"""Unit tests for Alyx REST queries using the AlyxClient.rest method."""
from pathlib import Path
import unittest
import unittest.mock
import random
import string
from uuid import UUID
import io
from logging import WARNING
import numpy as np
from one.webclient import AlyxClient, _PaginatedResponse
from one.tests import TEST_DB_1, OFFLINE_ONLY
[docs]
@unittest.skipIf(OFFLINE_ONLY, 'online only test')
class TestREST(unittest.TestCase):
"""Tests for AlyxClient.rest method and remote Alyx REST interactions."""
EID = 'cf264653-2deb-44cb-aa84-89b82507028a'
EID_EPHYS = 'b1c968ad-4874-468d-b2e4-5ffa9b9964e9'
alyx = None
[docs]
@classmethod
def setUpClass(cls) -> None:
cls.alyx = AlyxClient(**TEST_DB_1)
[docs]
def test_id_django_fields(self):
"""
Test.
The id and django kwargs have a specific meaning when using the list actions
on endpoints that implement the Alyx base REST framework. On endpoints that
do not implement the base REST framework, such as dataset_types and revisions,
the id/django args should yield an error.
:return:
"""
with self.assertWarns(UserWarning):
self.alyx.rest('revisions', 'list', id='boum')
with self.assertWarns(UserWarning):
self.alyx.rest('revisions', 'list', django='boum')
[docs]
def test_validation_list_fields(self):
# test raising an error when specifying a non-existent action
with self.assertRaises(ValueError) as cm:
self.alyx.rest('datasets', 'erase')
self.assertIn(
'Action "erase" for REST endpoint "datasets" does not exist', str(cm.exception))
# test raising an error when specifying a non-existent field
with self.assertWarns(UserWarning) as cm:
self.alyx.rest('insertions', 'list', djangodd='boum')
self.assertIn("Unsupported fields '{'djangodd'}' in query parameters", str(cm.warning))
[docs]
def test_paginated_request(self):
"""Check that paginated response object is returned upon making large queries."""
rep = self.alyx.rest('datasets', 'list')
self.assertTrue(isinstance(rep, _PaginatedResponse))
self.assertTrue(len(rep) > 250)
# This fails when new records are added/removed from the remote db while iterating
# self.assertTrue(len([_ for _ in rep]) == len(rep))
# Test what happens when list changes between paginated requests
name = '0A' + str(random.randint(0, 10000))
# Add subject between calls
rep = self.alyx.rest('subjects', 'list', limit=5, no_cache=True)
s = self.alyx.rest('subjects', 'create', data={'nickname': name, 'lab': 'cortexlab'})
self.addCleanup(self.alyx.rest, 'subjects', 'delete', id=s['nickname'])
with self.assertWarns(RuntimeWarning):
_ = rep[10]
[docs]
def test_generic_request(self):
"""Test AlyxClient.get method."""
a = self.alyx.get('/labs')
b = self.alyx.get('labs')
self.assertEqual(a, b)
[docs]
def test_rest_endpoint_write(self):
"""Test create and delete actions in AlyxClient.rest method."""
# test object creation and deletion with weighings
wa = {'subject': 'flowers',
'date_time': '2018-06-30T12:34:57',
'weight': 22.2,
'user': 'olivier'
}
a = self.alyx.rest('weighings', 'create', data=wa)
b = self.alyx.rest('weighings', 'read', id=a['url'])
self.assertEqual(a, b)
self.alyx.rest('weighings', 'delete', id=a['url'])
# test patch object with subjects
data = {'birth_date': '2018-04-01',
'death_date': '2018-09-10'}
sub = self.alyx.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.alyx.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):
"""Test list and read actions in AlyxClient.rest method."""
# tests that non-existing endpoints /actions are caught properly
with self.assertRaises(ValueError):
self.alyx.rest(url='turlu', action='create')
with self.assertRaises(ValueError):
self.alyx.rest(url='sessions', action='turlu')
# test with labs : get
a = self.alyx.rest('labs', 'list')
self.assertTrue(len(a) >= 3)
b = self.alyx.rest('/labs', 'list')
self.assertTrue(a == b)
# test with labs: read
c = self.alyx.rest('labs', 'read', 'mainenlab')
self.assertTrue([lab for lab in a if lab['name'] == 'mainenlab'][0] == c)
# test read with UUID object
dset = self.alyx.rest('datasets', 'read', id=UUID('738eca6f-d437-40d6-a9b8-a3f4cbbfbff7'))
self.assertEqual(dset['name'], '_iblrig_videoCodeFiles.raw.zip')
# Test with full URL
d = self.alyx.rest(
'labs', 'read',
f'{TEST_DB_1["base_url"]}/labs/mainenlab')
self.assertEqual(c, d)
# test a more complex endpoint with a filter and a selection
sub = self.alyx.rest('subjects/flowers', 'list')
sub1 = self.alyx.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.alyx.rest('/subjects?nickname=flowers')
self.assertEqual(sub1, sub2)
[docs]
def test_rest_all_actions(self):
"""Test for AlyxClient.rest method using subjects endpoint."""
# randint reduces conflicts with parallel tests
nickname = f'foobar_{random.randint(0, 10000)}'
newsub = {
'nickname': nickname,
'responsible_user': 'olivier',
'birth_date': '2019-06-15',
'death_date': None,
'lab': 'cortexlab',
}
# look for the subject, create it if necessary
sub = self.alyx.get(f'/subjects?&nickname={nickname}', expires=True)
if sub:
self.alyx.rest('subjects', 'delete', id=nickname)
self.alyx.rest('subjects', 'create', data=newsub)
# partial update (with hyphen correction) and full update
newsub = self.alyx.rest('subjects', 'partial-update',
id=nickname, data={'description': 'hey'})
self.assertEqual(newsub['description'], 'hey')
newsub['description'] = 'hoy'
newsub = self.alyx.rest('subjects', 'update', id=nickname, data=newsub)
self.assertEqual(newsub['description'], 'hoy')
# read
newsub_ = self.alyx.rest('subjects', 'read', id=nickname)
self.assertEqual(newsub, newsub_)
# list with filter
sub = self.alyx.rest('subjects', 'list', nickname=nickname)
self.assertEqual(sub[0]['nickname'], newsub['nickname'])
self.assertTrue(len(sub) == 1)
# Check that django=None is removed from kwargs
with unittest.mock.patch.object(self.alyx, 'get') as mock_get:
self.alyx.rest('subjects', 'list', django=None, lab='foo')
mock_get.assert_called_once_with('/subjects?lab=foo', clobber=False, expires=False)
# delete
self.alyx.rest('subjects', 'delete', id=nickname)
self.alyx.clear_rest_cache() # Make sure we hit db
sub = self.alyx.get(f'/subjects?&nickname={nickname}', expires=True)
self.assertFalse(sub)
[docs]
def test_endpoints_docs(self):
"""Test for AlyxClient.list_endpoints method and AlyxClient.rest."""
# Test endpoint documentation and validation
endpoints = self.alyx.rest_schemes.endpoints
self.assertTrue('auth-token' not in endpoints)
# Check that calling rest method with no args prints endpoints
with unittest.mock.patch('sys.stdout', new_callable=io.StringIO) as stdout:
self.alyx.rest()
self.assertTrue(k in stdout.getvalue() for k in endpoints)
# Same but with no action
with unittest.mock.patch('sys.stdout', new_callable=io.StringIO) as stdout:
self.assertIsNone(self.alyx.rest('sessions'))
actions = self.alyx.rest_schemes.actions('sessions')
self.assertTrue(all(k in stdout.getvalue() for k in actions))
expected = "['create', 'delete', 'list', 'partial_update', 'read', 'update']\n"
self.assertEqual(expected, stdout.getvalue()[:65])
# Check raises when endpoint invalid
self.assertRaises(ValueError, self.alyx.rest, 'foobar')
# Check logs warning when no id provided
with self.assertLogs('one.webclient', WARNING):
self.assertIsNone(self.alyx.rest('sessions', 'read'))
# Check logs warning when creating record with missing data
with self.assertLogs('one.webclient', WARNING):
self.assertIsNone(self.alyx.rest('sessions', 'create'))
with self.assertRaises(ValueError) as e:
self.alyx.json_field_write('foobar')
self.assertTrue(k in str(e.exception) for k in endpoints)
"""Specific Alyx REST endpoint tests"""
[docs]
def test_water_restriction(self):
"""Test listing water-restriction endpoint.
Examples of how to list all water restrictions and water-restriction for a given
subject.
"""
# get all the water restrictions from start
all_wr = self.alyx.rest('water-restriction', 'list')
# 2 different ways to get water restrictions for one subject
wr_sub2 = self.alyx.rest('water-restriction', 'list', subject='algernon') # recommended
# enforce test logic
expected = {'end_time', 'reference_weight', 'start_time', 'subject', 'water_type'}
self.assertTrue(expected >= set(all_wr[0].keys()))
self.assertTrue(len(all_wr) > len(wr_sub2))
[docs]
def test_list_pk_query(self):
"""Test REST list with id keyword argument.
It's a bit stupid but the REST endpoint can't forward a direct query of the uuid via
the pk keyword. The AlyxClient has already an id parameter, which on the list method
is used as a pk identifier. This special case is tested here.
"""
# Sessions returned sorted: take last session as new sessions constantly added and
# removed by parallel test runs
ses = self.alyx.rest('sessions', 'list')[-1]
eid = UUID(ses['url'][-36:]) # Should work with UUID object
ses_ = self.alyx.rest('sessions', 'list', id=eid)[-1]
self.assertEqual(ses, ses_)
# Check works with django query arg
query = f'start_time__date,{ses["start_time"][:10]}'
ses_ = self.alyx.rest('sessions', 'list', id=eid, django=query)[-1]
self.assertEqual(ses, ses_)
[docs]
def test_note_with_picture_upload(self):
"""Test adding session note with attached picture."""
my_note = {'user': 'olivier',
'content_type': 'session',
'object_id': self.EID,
'text': 'gnagnagna'}
png = Path(__file__).parent.joinpath('fixtures', 'test_img.png')
with open(png, 'rb') as img_file:
files = {'image': img_file}
ar_note = self.alyx.rest('notes', 'create', data=my_note, files=files)
self.assertTrue(len(ar_note['image']))
self.assertTrue(ar_note['content_type'] == 'actions.session')
self.alyx.rest('notes', 'delete', id=ar_note['id'])
[docs]
def test_channels(self):
"""Test creation of insertion, trajectory and channels."""
# need to build insertion + trajectory + channels to test the serialization of a
# record array in the channel endpoint
name = ''.join(random.choices(string.ascii_letters, k=5))
# Find any existing insertions with this name and delete (unlikely to find any)
probe_insertions = self.alyx.rest('insertions', 'list',
session=self.EID_EPHYS, name=name, no_cache=True)
for pi in probe_insertions:
self.alyx.rest('insertions', 'delete', pi['id'])
# Create new insertion with this name and add teardown hook to delete it
probe_insertion = self.alyx.rest(
'insertions', 'create', data={'session': self.EID_EPHYS, 'name': name})
self.addCleanup(self.alyx.rest, 'insertions', 'delete', id=probe_insertion['id'])
trajectory = self.alyx.rest('trajectories', 'create', data={
'probe_insertion': probe_insertion['id'],
'chronic_insertion': None,
'x': 1500,
'y': -2000,
'z': 0,
'depth': 4500,
'phi': 0,
'theta': 0,
'provenance': 'Histology track',
})
channel_records = []
for _ in np.arange(3):
channel_records.append({
'x': np.random.randint(-2000, 2000),
'y': np.random.randint(-2000, 2000),
'z': np.random.randint(-2000, 2000),
'axial': np.random.rand() * 800,
'lateral': np.random.rand() * 8,
'brain_region': 889,
'trajectory_estimate': trajectory['id']
})
channels = self.alyx.rest('channels', 'create', data=channel_records)
self.assertTrue(len(channels) == 3)
if __name__ == '__main__':
unittest.main(exit=False)