Source code for oneibl.tests.test_alyxclient

import unittest
import numpy as np
import os
import oneibl.webclient as wc
import oneibl.params
import tempfile
import shutil

from ibllib.io import hashfile

par = oneibl.params.get()

# Init connection to the database
ac = wc.AlyxClient(
    username='test_user', password='TapetesBloc18',
    base_url='https://test.alyx.internationalbrainlab.org')


[docs]class TestSingletonPattern(unittest.TestCase):
[docs] def setUp(self): self.ac = ac self.sameac = wc.AlyxClient( username='test_user', password='TapetesBloc18', base_url='https://test.alyx.internationalbrainlab.org') self.sameac2 = wc.AlyxClient( username='test_user', password='TapetesBloc18', base_url='https://test.alyx.internationalbrainlab.org')
[docs] def test_multiple_singletons(self): self.assertTrue(id(self.ac) == id(self.sameac)) self.assertTrue(id(self.ac) == id(self.sameac2))
[docs]class TestJsonFieldMethods(unittest.TestCase):
[docs] def setUp(self): self.ac = ac self.eid1 = '242f2929-faaf-4e7c-ae3f-4a935c6d8da5' self.eid2 = 'dfe99506-b873-45db-bc93-731f9362e304' self.endpoint = "sessions" self.field_name = "extended_qc" self.data_dict = {'some': 0, 'data': 1} self.ac.json_field_delete(self.endpoint, self.eid1, self.field_name) self.ac.json_field_delete(self.endpoint, self.eid2, self.field_name) self.eid1_eqc = None self.eid2_eqc = None
def _json_field_write(self): written1 = self.ac.json_field_write( self.endpoint, self.eid1, self.field_name, self.data_dict ) written2 = self.ac.json_field_write( self.endpoint, self.eid2, self.field_name, self.data_dict ) self.assertTrue(written1 == written2) self.assertTrue(written1 == self.data_dict) sess_dict = self.ac.rest(self.endpoint, 'list', extended_qc='some__lt,0.5') self.assertTrue(len(sess_dict) == 2) def _json_field_update(self): modified = self.ac.json_field_update( self.endpoint, self.eid1, self.field_name, {'some': 0.6} ) self.assertTrue('data' in modified) self.assertTrue('some' in modified) self.assertTrue(len(self.ac.rest(self.endpoint, 'list', extended_qc='some__lt,0.5')) == 1) def _json_field_remove_key(self): pre_delete = self.ac.rest(self.endpoint, 'list', extended_qc='data__gte,0.5') self.assertTrue(len(pre_delete) == 2) deleted = self.ac.json_field_remove_key(self.endpoint, self.eid2, self.field_name, 'data') self.assertTrue('data' not in deleted) post_delete = self.ac.rest(self.endpoint, 'list', extended_qc='data__gte,0.5') self.assertTrue(len(post_delete) == 1) def _json_field_delete(self): deleted = self.ac.json_field_delete(self.endpoint, self.eid2, self.field_name) self.assertTrue(deleted is None) self.assertTrue(len(self.ac.rest(self.endpoint, 'list', extended_qc='data__gte,0.5')) == 1)
[docs] def test_json_methods(self): self._json_field_write() self._json_field_update() self._json_field_remove_key() self._json_field_delete()
[docs] def tearDown(self): # Delete any dict created by this test for x in self.ac.rest(self.endpoint, 'list', extended_qc='some__lt,0.5'): self.ac.json_field_delete(self.endpoint, x['url'][-36:], self.field_name) # Restore whatever was there in the first place self.ac.json_field_write(self.endpoint, self.eid1, self.field_name, self.eid1_eqc) self.ac.json_field_write(self.endpoint, self.eid2, self.field_name, self.eid2_eqc)
[docs]class TestDownloadHTTP(unittest.TestCase):
[docs] def setUp(self): self.ac = ac self.test_data_uuid = '3ddd45be-7d24-4fc7-9dd3-a98717342af6'
[docs] def test_paginated_request(self): rep = self.ac.rest('datasets', 'list') self.assertTrue(isinstance(rep, oneibl.webclient._PaginatedResponse)) self.assertTrue(len(rep) > 250) self.assertTrue(len([d['hash'] for d in rep]) == len(rep))
[docs] def test_generic_request(self): a = self.ac.get('/labs') b = self.ac.get('labs') self.assertEqual(a, b)
[docs] def test_rest_endpoint_write(self): # test object creation and deletion with weighings wa = {'subject': 'flowers', 'date_time': '2018-06-30T12:34:57', 'weight': 22.2, 'user': 'olivier' } a = self.ac.rest('weighings', 'create', data=wa) b = self.ac.rest('weighings', 'read', id=a['url']) self.assertEqual(a, b) self.ac.rest('weighings', 'delete', id=a['url']) # test patch object with subjects sub = self.ac.rest('subjects', 'read', id='flowers') data = {'birth_date': '2018-04-01', 'death_date': '2018-09-10'} sub = self.ac.rest('subjects', 'partial_update', id='flowers', data=data) self.assertEqual(sub['birth_date'], data['birth_date']) self.assertEqual(sub['death_date'], data['death_date']) data = {'birth_date': '2018-04-02', 'death_date': '2018-09-09'} sub = self.ac.rest('subjects', 'partial_update', id='flowers', data=data) self.assertEqual(sub['birth_date'], data['birth_date']) self.assertEqual(sub['death_date'], data['death_date'])
[docs] def test_rest_endpoint_read_only(self): # tests that non-existing endpoints /actions are caught properly with self.assertRaises(ValueError): self.ac.rest(url='turlu', action='create') with self.assertRaises(ValueError): self.ac.rest(url='sessions', action='turlu') # test with labs : get a = self.ac.rest('labs', 'list') self.assertTrue(len(a) >= 3) b = self.ac.rest('/labs', 'list') self.assertTrue(a == b) # test with labs: read c = self.ac.rest('labs', 'read', 'mainenlab') self.assertTrue([lab for lab in a if lab['name'] == 'mainenlab'][0] == c) d = self.ac.rest( 'labs', 'read', 'https://test.alyx.internationalbrainlab.org/labs/mainenlab') self.assertEqual(c, d) # test a more complex endpoint with a filter and a selection sub = self.ac.rest('subjects/flowers', 'list') sub1 = self.ac.rest('subjects?nickname=flowers', 'list') self.assertTrue(len(sub1) == 1) self.assertEqual(sub['nickname'], sub1[0]['nickname']) # also make sure the action is overriden on a filter query sub2 = self.ac.rest('/subjects?nickname=flowers') self.assertEqual(sub1, sub2)
[docs] def test_download_datasets_with_api(self): ac = self.ac # easier to debug in console test_data_uuid = self.test_data_uuid cache_dir = tempfile.mkdtemp() # Test 1: empty dir, dict mode dset = ac.get('/datasets/' + test_data_uuid) url = wc.dataset_record_to_url(dset) file_name = wc.http_download_file_list(url, username=par.HTTP_DATA_SERVER_LOGIN, password=par.HTTP_DATA_SERVER_PWD, cache_dir=cache_dir) # Test 2: empty dir, list mode dset = ac.get('/datasets?id=' + test_data_uuid) url = wc.dataset_record_to_url(dset) file_name = wc.http_download_file_list(url, username=par.HTTP_DATA_SERVER_LOGIN, password=par.HTTP_DATA_SERVER_PWD, cache_dir=cache_dir) self.assertTrue(os.path.isfile(file_name[0])) shutil.rmtree(cache_dir)
[docs] def test_download_datasets(self): # test downloading a single file full_link_to_file = r'http://ibl.flatironinstitute.org/mainenlab/Subjects/clns0730'\ '/2018-08-24/1/licks.times.51852a2f-c76e-4c0c-95cb-9c7ba54be0f9.npy' file_name, md5 = wc.http_download_file(full_link_to_file, username=par.HTTP_DATA_SERVER_LOGIN, password=par.HTTP_DATA_SERVER_PWD, return_md5=True, clobber=True) a = np.load(file_name) self.assertTrue(hashfile.md5(file_name) == md5) self.assertTrue(len(a) > 0) # test downloading a list of files links = [r'http://ibl.flatironinstitute.org/mainenlab/Subjects/clns0730' '/2018-08-24/1/licks.times.51852a2f-c76e-4c0c-95cb-9c7ba54be0f9.npy', r'http://ibl.flatironinstitute.org/mainenlab/Subjects/clns0730' '/2018-08-24/1/probes.sitePositions.3ddd45be-7d24-4fc7-9dd3-a98717342af6.npy' ] file_list = wc.http_download_file_list(links, username=par.HTTP_DATA_SERVER_LOGIN, password=par.HTTP_DATA_SERVER_PWD) a = np.load(file_list[0]) b = np.load(file_list[1]) self.assertTrue(len(a) > 0) self.assertTrue(len(b) > 0)
[docs] def test_rest_all_actions(self): newsub = { 'nickname': 'tutu', 'responsible_user': 'olivier', 'birth_date': '2019-06-15', 'death_date': None, 'lab': 'cortexlab', } # look for the subject, create it if necessary sub = self.ac.rest('subjects', 'list', nickname='tutu') if sub: self.ac.rest('subjects', 'delete', id='tutu') newsub = self.ac.rest('subjects', 'create', data=newsub) # partial update and full update newsub = self.ac.rest('subjects', 'partial_update', id='tutu', data={'description': 'hey'}) self.assertEqual(newsub['description'], 'hey') newsub['description'] = 'hoy' newsub = self.ac.rest('subjects', 'update', id='tutu', data=newsub) self.assertEqual(newsub['description'], 'hoy') # read newsub_ = self.ac.rest('subjects', 'read', id='tutu') self.assertEqual(newsub, newsub_) # list with filter sub = self.ac.rest('subjects', 'list', nickname='tutu') self.assertEqual(sub[0]['nickname'], newsub['nickname']) self.assertTrue(len(sub) == 1) # delete self.ac.rest('subjects', 'delete', id='tutu') sub = self.ac.rest('subjects', 'list', nickname='tutu') self.assertFalse(sub)
if __name__ == '__main__': unittest.main(exit=False, verbosity=2)