FAQ | This is a LIVE service | Changelog

Commit 2b1c6ded authored by Robin Goodall's avatar Robin Goodall 💬
Browse files

student_api tests

parent 76494981
......@@ -143,11 +143,15 @@ def fetch_all_students(
"""
next_url = urllib.parse.urljoin(STUDENT_API_ROOT, 'students')
first = True
while next_url is not None:
LOG.info('Fetching %s...', next_url)
r = session.get(next_url)
r.raise_for_status()
data = r.json()
for s in data.get('results', []):
if first:
first = False
LOG.info('%s', s)
yield Student.parse_obj(s)
next_url = data.get('next')
from typing import Dict
from contextlib import contextmanager
from tempfile import NamedTemporaryFile
import os
# Example Student Records institution value
SR_INST_VALUE = "SRID1"
@contextmanager
def temp_password_file(pwd: str):
......@@ -31,10 +35,11 @@ class MockResponse():
class MockSession():
def __init__(self, url: str, data: dict):
self.url = url
self.data = data
def __init__(self, responses: Dict[str, Dict]):
self.responses = responses
self.urls_got = set()
def get(self, url: str) -> MockResponse:
assert url == self.url
return MockResponse(self.data)
assert url in self.responses.keys()
self.urls_got.add(url)
return MockResponse(self.responses[url])
from typing import List, Dict, Optional, Tuple, Union
import faker
from identitylib.identifiers import IdentifierSchemes
from lookupsync.student_api import ACADEMIC_CAREER_MAPPING
from . import SR_INST_VALUE
class StudentProvider(faker.providers.BaseProvider):
"""
A faker provider for student data as provided by Student API
"""
def student(self, **kwargs) -> Dict[str, Union[str, Dict]]:
"""
A single fake student, overriding properties as necessary
"""
return {
k: kwargs.get(k) or getattr(self, k)(**kwargs)
for k in ['namePrefixes', 'surname', 'forenames', 'affiliations', 'identifiers']
}
def namePrefixes(self, **kwargs) -> str:
return self.generator.prefix().rstrip('.')
def surname(self, **kwargs) -> str:
return self.generator.last_name()
def forenames(self, **kwargs) -> str:
return self.generator.first_name() + (
'' if self.generator.boolean(chance_of_getting_true=25)
else (' ' + self.generator.first_name())
)
def affiliations(self, num_affiliations=1, **kwargs) -> List[Dict[str, str]]:
return [self.affiliation(**kwargs) for _ in range(num_affiliations)]
def identifiers(self, num_identifiers=1, **kwargs) -> List[Dict[str, str]]:
return [self.identifier(**kwargs) for _ in range(num_identifiers)]
def affiliation(self, **kwargs) -> Dict[str, str]:
value, scheme = (kwargs.get('affiliation_id') or self.affiliation_id(**kwargs)).split('@')
start, end = kwargs.get('affiliation_period') or self.affiliation_period(**kwargs)
return {
'value': value,
'scheme': scheme,
'start': start,
'end': end,
'status': kwargs.get('status') or self.affiliation_status(**kwargs),
}
def affiliation_id(self, **kwargs) -> str:
# Provides example affiliation (in test mapping) with expected scheme
return f'{SR_INST_VALUE}@{IdentifierSchemes.STUDENT_INSTITUTION}'
def affiliation_period(self, **kwargs) -> Tuple[Optional[str], Optional[str]]:
# Unspecified start and end dates
return (None, None)
def affiliation_status(self, **kwargs) -> str:
return self.random_element(ACADEMIC_CAREER_MAPPING.keys())
def identifier(self, **kwargs) -> Dict[str, str]:
value, scheme = (kwargs.get('identifier_id') or self.identifier_id(**kwargs)).split('@')
return {
'value': value,
'scheme': scheme,
}
def identifier_id(self, **kwargs) -> str:
# Provides random person with expected scheme
return f'{self.random_int(10000000, 99999999)}@{IdentifierSchemes.USN}'
from unittest import TestCase
import logging
from . import MockSession
from . import MockSession, SR_INST_VALUE
from lookupsync.inst_mapping import fetch_inst_mapping, INST_MAPPING_API_ROOT
INST_MAPPING_RESPONSE = {
......@@ -12,7 +12,7 @@ INST_MAPPING_RESPONSE = {
# Duplicate of "instid" - not actually used in sync
"FOO@insts.lookup.cam.ac.uk",
# current scheme
"SRID1@institution.v1.student-records.university.identifiers.cam.ac.uk",
f"{SR_INST_VALUE}@institution.v1.student-records.university.identifiers.cam.ac.uk",
# depreciated but still valid id
"SRID2@institution.v1.student.university.identifiers.cam.ac.uk",
# ignored id but using valid scheme
......@@ -36,7 +36,7 @@ INST_MAPPING_RESPONSE = {
]
}
INST_MAPPING_EXPECTED = {
'SRID1': 'FOO',
SR_INST_VALUE: 'FOO',
'SRID2': 'FOO',
'SRID100': 'BAR',
'SRID200': 'BAZ',
......@@ -52,7 +52,7 @@ class InstMappingTest(TestCase):
institution ids to Lookup instids
"""
session = MockSession(INST_MAPPING_API_ROOT, INST_MAPPING_RESPONSE)
session = MockSession({INST_MAPPING_API_ROOT: INST_MAPPING_RESPONSE})
with self.assertLogs(level=logging.WARNING) as captured:
inst_map = fetch_inst_mapping(session)
......
from typing import List, Dict
import logging
from unittest import TestCase
from urllib.parse import urljoin
import faker
from identitylib.identifiers import IdentifierSchemes
from . import MockSession, SR_INST_VALUE
from lookupsync.tests.fakeproviders import StudentProvider
from lookupsync.student_api import (
fetch_all_students, get_students_by_group,
STUDENT_API_ROOT, ACADEMIC_CAREER_MAPPING, Student)
from lookupsync.lookup import group_name
STUDENT_API_INITIAL_PAGE_URL = urljoin(STUDENT_API_ROOT, 'students')
INST_MAPPING = {
SR_INST_VALUE: 'FOO',
'SRID2': 'FOO',
'SRID3': 'BAR',
'SRID4': 'BAZ',
}
# Typical response from Student Records API
# {
# 'results': [
# {
# 'identifiers': [
# {
# 'value': '123456789',
# 'scheme': 'person.v1.student-records.university.identifiers.cam.ac.uk'
# }
# ],
# 'namePrefixes': 'Mr',
# 'surname': 'Smith',
# 'forenames': 'John George',
# 'dateOfBirth': '2000-01-02',
# 'affiliations': [
# {
# 'value': 'EDM2PSY',
# 'status': 'PGRD',
# 'scheme': 'academic-plan.v1.student-records.university.identifiers.cam.ac.uk',
# 'start': '2021-10-01',
# 'end': '2022-06-18'
# }, {
# 'value': 'W',
# 'status': 'PGRD',
# 'scheme': 'institution.v1.student-records.university.identifiers.cam.ac.uk',
# 'start': '2021-10-01',
# 'end': '2022-06-18'
# }, {
# 'value': 'EF',
# 'status': 'PGRD',
# 'scheme': 'institution.v1.student-records.university.identifiers.cam.ac.uk',
# 'start': '2021-10-01',
# 'end': '2022-06-18'
# }
# ]
# }
# ],
# 'next': '....?cursor=FOO',
# }
class StudentAPITest(TestCase):
def setUp(self):
self.fake = faker.Faker()
self.fake.seed_instance(0xdeadbeef)
self.fake.add_provider(StudentProvider)
def test_fetch_all_students(self):
"""
fetch_all_students gets paged responses from the Student API yielding each
student as a Student instance.
"""
# Create 100 random students and split them in to pages
NUMBER_STUDENTS = 100
PAGE_SIZE = 29
fake_students = [self.fake.student() for _ in range(NUMBER_STUDENTS)]
pages = [fake_students[i:i + PAGE_SIZE] for i in range(0, len(fake_students), PAGE_SIZE)]
# Create a dict of requests and responses for MockSession to provide
responses = {}
all_urls = set()
for idx, page in enumerate(pages):
url = STUDENT_API_INITIAL_PAGE_URL + (
'' if idx == 0 else f'?cursor=PAGE{idx}'
)
all_urls.add(url)
next = (
None if idx == len(pages)-1
else f'{STUDENT_API_INITIAL_PAGE_URL}?cursor=PAGE{idx+1}'
)
responses[url] = {
'results': [s for s in page],
'next': next,
}
self.assertEqual(len(all_urls), len(pages))
session = MockSession(responses)
# Fetch all students
all_students = list(fetch_all_students(session))
# All pages requested
self.assertEqual(session.urls_got, all_urls)
# Result matches expectations
self.assertEqual(all_students, [
Student.parse_obj(s) for s in fake_students
])
def _students_to_responses(self, students: List[Dict]) -> Dict[str, Dict]:
"""
Take a list of students and return a single page response for the initial
page request
"""
return {
STUDENT_API_INITIAL_PAGE_URL: {
'results': [s for s in students],
'next': None,
}
}
def test_get_students_by_group(self):
"""
Students with a single affiliation and USN identifier all get returned
"""
# Provider gives us students like this by default
NUMBER_STUDENTS = 100
while True:
fake_students = [self.fake.student() for _ in range(NUMBER_STUDENTS)]
# Make sure at least one affiliation per expected status
statuses = {s['affiliations'][0]['status'] for s in fake_students}
if set(statuses) == set(ACADEMIC_CAREER_MAPPING.keys()):
break
# get students by groups with list of fake students in mock response
session = MockSession(self._students_to_responses(fake_students))
students_by_group = get_students_by_group(session, INST_MAPPING)
# Expect groups for all status at single lookup inst match keys of result
lookup_inst = INST_MAPPING.get(SR_INST_VALUE)
expected_groups = {
group_name(lookup_inst, status)
for status in ACADEMIC_CAREER_MAPPING.values()
}
self.assertEqual(expected_groups, set(students_by_group.keys()))
# Expect all students to be returned (being only in a single group)
count = sum([len(s) for s in students_by_group.values()])
self.assertEqual(count, NUMBER_STUDENTS)
def test_get_students_by_group_multiple_affiliations(self):
"""
Students with multiple different affiliations get assigned to all matching groups
"""
# Single student
STUDENT_USN = '123456789'
identifier_id = f'{STUDENT_USN}@{IdentifierSchemes.USN}'
# in multiple distinct lookup institutions
AFFILIATION_ID_VALUES = {'SRID1', 'SRID3', 'SRID4'}
lookup_id_values = {INST_MAPPING.get(v) for v in AFFILIATION_ID_VALUES}
affiliations = [
self.fake.affiliation(
affiliation_id=f'{value}@{IdentifierSchemes.STUDENT_INSTITUTION}',
status='UGRD', # as an undergraduate
)
for value in AFFILIATION_ID_VALUES
]
fake_student = self.fake.student(
identifier_id=identifier_id,
affiliations=affiliations
)
# get students by groups with just fake student in mock response
session = MockSession(self._students_to_responses([fake_student]))
students_by_group = get_students_by_group(session, INST_MAPPING)
# Expect groups for all status at single lookup inst match keys of result
expected_groups = {
group_name(lookup_id, 'ug')
for lookup_id in lookup_id_values
}
self.assertEqual(expected_groups, set(students_by_group.keys()))
# The student is the single member of each list
for group, students in students_by_group.items():
self.assertEqual(len(students), 1)
self.assertEqual(students, {STUDENT_USN})
def test_get_students_by_group_depreciated(self):
"""
Students with depreciated but valid Student Records inst or person id schemes still get
mapped to appropriate group
"""
# Single student with depreciated USN scheme
STUDENT_USN = '123456789'
identifier_id = f'{STUDENT_USN}@person.camsis.identifiers.admin.cam.ac.uk'
# and affiliation with depreciated institution scheme
AFFILIATION_VALUE = 'SRID3'
affiliation_id = (
f'{AFFILIATION_VALUE}@institution.v1.student.university.identifiers.cam.ac.uk')
lookup_id = INST_MAPPING.get(AFFILIATION_VALUE)
fake_student = self.fake.student(
identifier_id=identifier_id,
affiliation_id=affiliation_id,
status='PGRD', # as a postgraduate
)
# get students by groups with just fake student in mock response
session = MockSession(self._students_to_responses([fake_student]))
students_by_group = get_students_by_group(session, INST_MAPPING)
# Expect just single lookup group with single matching student
expected_group = group_name(lookup_id, 'pg')
self.assertEqual({expected_group: {STUDENT_USN}}, students_by_group)
def test_get_students_by_group_no_aff(self):
"""
Students with no appropriate affiliations don't appear in any group
"""
# Single student with USN but no affiliations
STUDENT_USN = '123456789'
identifier_id = f'{STUDENT_USN}@{IdentifierSchemes.USN}'
fake_student = self.fake.student(
identifier_id=identifier_id,
affiliation_id=f'ADMIN@{IdentifierSchemes.HR_INSTITUTION}', # not Student Records inst
)
# get students by groups with just fake student in mock response
session = MockSession(self._students_to_responses([fake_student]))
students_by_group = get_students_by_group(session, INST_MAPPING)
# Expect empty result
self.assertEqual({}, students_by_group)
def test_get_students_by_group_invalid_aff(self):
"""
Students with affiliations with invalid scheme don't appear as group isn't even created
"""
# Single student with USN but only invalid affiliation
STUDENT_USN = '123456789'
identifier_id = f'{STUDENT_USN}@{IdentifierSchemes.USN}'
fake_student = self.fake.student(
identifier_id=identifier_id,
affiliation_id='SRID1@invalid.inst.id.scheme',
)
# get students by groups with just fake student in mock response
session = MockSession(self._students_to_responses([fake_student]))
students_by_group = get_students_by_group(session, INST_MAPPING)
# Expect empty result
self.assertEqual({}, students_by_group)
def test_get_students_by_group_no_usn(self):
"""
Students with no USN don't appear in any group though group gets created
"""
# Single student with affiliation but no USN
AFFILIATION_VALUE = 'SRID3'
affiliation_id = f'{AFFILIATION_VALUE}@{IdentifierSchemes.STUDENT_INSTITUTION}'
lookup_id = INST_MAPPING.get(AFFILIATION_VALUE)
fake_student = self.fake.student(
identifier_id=f'abc123@{IdentifierSchemes.CRSID}', # CRSid instead of USN
affiliation_id=affiliation_id,
status='PGRD', # as a postgraduate
)
# get students by groups with just fake student in mock response
session = MockSession(self._students_to_responses([fake_student]))
students_by_group = get_students_by_group(session, INST_MAPPING)
# Expect result with single group but no students
expected_group = group_name(lookup_id, 'pg')
self.assertEqual({expected_group: set()}, students_by_group)
def test_get_students_by_group_invalid_usn(self):
"""
Students with invalid person identifier scheme don't appear in matching group
"""
# Single student with invalid person identifier scheme
STUDENT_USN = '123456789'
AFFILIATION_VALUE = 'SRID3'
affiliation_id = f'{AFFILIATION_VALUE}@{IdentifierSchemes.STUDENT_INSTITUTION}'
lookup_id = INST_MAPPING.get(AFFILIATION_VALUE)
fake_student = self.fake.student(
identifier_id=f'{STUDENT_USN}@invalid.person.id.scheme',
affiliation_id=affiliation_id,
status='PGRD', # as a postgraduate
)
# get students by groups with just fake student in mock response
session = MockSession(self._students_to_responses([fake_student]))
students_by_group = get_students_by_group(session, INST_MAPPING)
# Expect result with single group but no students
expected_group = group_name(lookup_id, 'pg')
self.assertEqual({expected_group: set()}, students_by_group)
def test_get_students_by_group_periods(self):
"""
Students with current, expired and yet to start affiliations only appear in groups for
the current affiliation.
"""
# Student with multiple affiliation
STUDENT_USN = '123456789'
identifier_id = f'{STUDENT_USN}@{IdentifierSchemes.USN}'
AFFILIATION_ID_VALUES = ['SRID1', 'SRID3', 'SRID4']
lookup_id_values = [INST_MAPPING.get(v) for v in AFFILIATION_ID_VALUES]
# but only one current
PAST_DATE = self.fake.past_date().strftime('%Y-%m-%d')
FUTURE_DATE = self.fake.future_date().strftime('%Y-%m-%d')
# first is in past, second is current, last is in future
periods = [(None, PAST_DATE), (PAST_DATE, FUTURE_DATE), (FUTURE_DATE, None)]
affiliations = [
self.fake.affiliation(
affiliation_id=f'{value}@{IdentifierSchemes.STUDENT_INSTITUTION}',
status='UGRD', # as an undergraduate
affiliation_period=periods[idx],
)
for idx, value in enumerate(AFFILIATION_ID_VALUES)
]
fake_student = self.fake.student(
identifier_id=identifier_id,
affiliations=affiliations
)
# get students by groups with just fake student in mock response
session = MockSession(self._students_to_responses([fake_student]))
students_by_group = get_students_by_group(session, INST_MAPPING)
# Expect just one group (for current affiliation) with the student
expected_group = group_name(lookup_id_values[1], 'ug')
self.assertEqual({expected_group: {STUDENT_USN}}, students_by_group)
def test_get_students_by_group_ignored(self):
"""
Students with unmappable affiliations or careers cause these to be logged
"""
# Student with multiple affiliation (last one unmappable)
STUDENT_USN = '123456789'
identifier_id = f'{STUDENT_USN}@{IdentifierSchemes.USN}'
AFFILIATION_ID_VALUES = ['SRID1', 'SRID3', 'BADAFF']
lookup_id_values = [INST_MAPPING.get(v) for v in AFFILIATION_ID_VALUES]
# and second affiliation with unmappable career
CAREERS = ['UGRD', 'BADCAR', 'PGRD']
career_mappings = [ACADEMIC_CAREER_MAPPING.get(c) for c in CAREERS]
affiliations = [
self.fake.affiliation(
affiliation_id=f'{value}@{IdentifierSchemes.STUDENT_INSTITUTION}',
status=CAREERS[idx],
)
for idx, value in enumerate(AFFILIATION_ID_VALUES)
]
fake_student = self.fake.student(
identifier_id=identifier_id,
affiliations=affiliations
)
# get students by groups with just fake student in mock response
session = MockSession(self._students_to_responses([fake_student]))
with self.assertLogs(level=logging.INFO) as captured:
students_by_group = get_students_by_group(session, INST_MAPPING)
# Result contains just mappable affiliation (the first)
expected_group = group_name(lookup_id_values[0], career_mappings[0])
self.assertEqual({expected_group: {STUDENT_USN}}, students_by_group)
log_messages = [r.getMessage() for r in captured.records]
# Warning given for ignored affiliation
try:
ignored_log_idx = log_messages.index('Ignored Affiliations:')
except ValueError:
self.fail('No ignored affiliations')
self.assertEqual('- BADAFF', log_messages[ignored_log_idx+1])
# Warning given for ignored career
try:
ignored_log_idx = log_messages.index('Ignored Academic Careers:')
except ValueError:
self.fail('No ignored careers')
self.assertEqual('- BADCAR', log_messages[ignored_log_idx+1])
......@@ -28,6 +28,7 @@ build_root={env:TOXINI_ARTEFACT_DIR:{toxinidir}/build}
deps=
-rrequirements.txt
coverage
faker
pytest
pytest-cov
# Which environment variables should be passed into the environment.
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment