generated from communitiesuk/funding-service-design-TEMPLATE
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added API routes and db tests for user application association
- Loading branch information
Kishan Patel
authored and
Kishan Patel
committed
Jun 21, 2024
1 parent
1765974
commit ef72b12
Showing
7 changed files
with
275 additions
and
12 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,14 +1,67 @@ | ||
def get_all_users_associated_with_application(): | ||
pass | ||
from db.queries.assessment_records.queries import create_user_application_association | ||
from db.queries.assessment_records.queries import get_user_application_associations | ||
from db.queries.assessment_records.queries import ( | ||
update_user_application_association as update_user_application_association_db, | ||
) | ||
from db.schemas.schemas import AllocationAssociationSchema | ||
from flask import abort | ||
from flask import current_app | ||
from flask import request | ||
|
||
|
||
def get_user_application_association(): | ||
pass | ||
def get_all_users_associated_with_application(application_id, active=None): | ||
associations = get_user_application_associations(application_id=application_id, active=active) | ||
if associations: | ||
serialiser = AllocationAssociationSchema() | ||
return serialiser.dump(associations, many=True) | ||
|
||
current_app.logger.error(f"Could not find any users associated with application {application_id}") | ||
abort(404) | ||
|
||
def add_user_application_association(): | ||
pass | ||
|
||
def get_user_application_association(application_id, user_id): | ||
association = get_user_application_associations(application_id=application_id, user_id=user_id) | ||
|
||
def update_user_application_association(): | ||
pass | ||
if association: | ||
serialiser = AllocationAssociationSchema() | ||
return serialiser.dump(association[0]) | ||
|
||
current_app.logger.error(f"Could not find association between {user_id} and application {application_id}") | ||
abort(404) | ||
|
||
|
||
def add_user_application_association(application_id, user_id): | ||
association = create_user_application_association(application_id=application_id, user_id=user_id) | ||
|
||
if association: | ||
serialiser = AllocationAssociationSchema() | ||
return serialiser.dump(association) | ||
|
||
current_app.logger.error(f"Could not create association between {user_id} and application {application_id}") | ||
abort(404) | ||
|
||
|
||
def update_user_application_association(application_id, user_id): | ||
args = request.get_json() | ||
if "active" not in args: | ||
abort(400) | ||
|
||
active = True if args.get("active").lower() == "true" else False | ||
association = update_user_application_association_db(application_id=application_id, user_id=user_id, active=active) | ||
|
||
if association: | ||
serialiser = AllocationAssociationSchema() | ||
return serialiser.dump(association) | ||
|
||
current_app.logger.error(f"Could not update association between {user_id} and application {application_id}") | ||
abort(404) | ||
|
||
|
||
def get_all_applications_associated_with_user(user_id, active=None): | ||
associations = get_user_application_associations(user_id=user_id, active=active) | ||
if associations: | ||
serialiser = AllocationAssociationSchema() | ||
return serialiser.dump(associations, many=True) | ||
|
||
current_app.logger.error(f"Could not find any applications associated with user {user_id}") | ||
abort(404) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,117 @@ | ||
import uuid | ||
from datetime import datetime | ||
from datetime import timezone | ||
|
||
import pytest | ||
from db.models.assessment_record.allocation_association import AllocationAssociation | ||
from db.queries.assessment_records.queries import create_user_application_association | ||
from db.queries.assessment_records.queries import get_user_application_associations | ||
from db.queries.assessment_records.queries import update_user_application_association | ||
from tests.conftest import test_input_data | ||
|
||
|
||
@pytest.mark.apps_to_insert([{**test_input_data[0]}]) | ||
def test_get_users_for_application(_db, seed_application_records): | ||
user_id_1 = str(uuid.uuid4()) | ||
user_id_2 = str(uuid.uuid4()) | ||
app_id = seed_application_records[0]["application_id"] | ||
allocation_association_1 = AllocationAssociation( | ||
user_id=user_id_1, | ||
application_id=app_id, | ||
active=True, | ||
log={datetime.now(tz=timezone.utc).isoformat(): "activated"}, | ||
) | ||
allocation_association_2 = AllocationAssociation( | ||
user_id=user_id_2, | ||
application_id=app_id, | ||
active=False, | ||
log={datetime.now(tz=timezone.utc).isoformat(): "deactivated"}, | ||
) | ||
_db.session.add(allocation_association_1) | ||
_db.session.commit() | ||
_db.session.add(allocation_association_2) | ||
_db.session.commit() | ||
users = get_user_application_associations(application_id=app_id) | ||
assert len(users) == 2 | ||
|
||
active_users = get_user_application_associations(application_id=app_id, active=True) | ||
assert len(active_users) == 1 | ||
assert str(active_users[0].user_id) == user_id_1 | ||
|
||
inactive_users = get_user_application_associations(application_id=app_id, active=False) | ||
assert len(inactive_users) == 1 | ||
assert str(inactive_users[0].user_id) == user_id_2 | ||
|
||
|
||
@pytest.mark.apps_to_insert([{**test_input_data[0]}, {**test_input_data[1]}]) | ||
def test_get_applications_for_user(_db, seed_application_records): | ||
user_id = str(uuid.uuid4()) | ||
app_id_1 = seed_application_records[0]["application_id"] | ||
app_id_2 = seed_application_records[1]["application_id"] | ||
allocation_association_1 = AllocationAssociation( | ||
user_id=user_id, | ||
application_id=app_id_1, | ||
active=True, | ||
log={datetime.now(tz=timezone.utc).isoformat(): "activated"}, | ||
) | ||
allocation_association_2 = AllocationAssociation( | ||
user_id=user_id, | ||
application_id=app_id_2, | ||
active=False, | ||
log={datetime.now(tz=timezone.utc).isoformat(): "deactivated"}, | ||
) | ||
_db.session.add(allocation_association_1) | ||
_db.session.commit() | ||
_db.session.add(allocation_association_2) | ||
_db.session.commit() | ||
applications = get_user_application_associations(user_id=user_id) | ||
assert len(applications) == 2 | ||
|
||
active_applications = get_user_application_associations(user_id=user_id, active=True) | ||
assert len(active_applications) == 1 | ||
assert str(active_applications[0].application_id) == app_id_1 | ||
|
||
inactive_applications = get_user_application_associations(user_id=user_id, active=False) | ||
assert len(inactive_applications) == 1 | ||
assert str(inactive_applications[0].application_id) == app_id_2 | ||
|
||
|
||
@pytest.mark.apps_to_insert([{**test_input_data[0]}]) | ||
def test_create_user_application_association(_db, seed_application_records): | ||
user_id = str(uuid.uuid4()) | ||
app_id = seed_application_records[0]["application_id"] | ||
new_association = create_user_application_association(app_id, user_id) | ||
assert new_association is not None | ||
assert str(new_association.application_id) == app_id | ||
assert str(new_association.user_id) == user_id | ||
assert new_association.active is True | ||
assert "activated" == list(new_association.log.values())[0] | ||
try: | ||
datetime.fromisoformat(list(new_association.log.keys())[0]) | ||
except ValueError: | ||
assert False, "log keys should be in isoformat" | ||
|
||
|
||
@pytest.mark.apps_to_insert([{**test_input_data[0]}]) | ||
def test_update_user_application_association(_db, seed_application_records): | ||
user_id = str(uuid.uuid4()) | ||
app_id = seed_application_records[0]["application_id"] | ||
create_user_application_association(app_id, user_id) | ||
updated_association = update_user_application_association(app_id, user_id, active=False) | ||
assert updated_association.active is False | ||
assert len(updated_association.log.keys()) == 2 | ||
logs = [(datetime.fromisoformat(key), value) for key, value in updated_association.log.items()] | ||
|
||
# The timestamp for deactivated should be after activated | ||
assert logs[0][1] == "activated" if logs[0][0] < logs[1][0] else "deactivated" | ||
assert logs[1][1] == "deactivated" if logs[0][0] < logs[1][0] else "activated" | ||
|
||
|
||
@pytest.mark.apps_to_insert([{**test_input_data[0]}]) | ||
def test_duplicate_user_application_association(_db, seed_application_records): | ||
user_id = str(uuid.uuid4()) | ||
app_id = seed_application_records[0]["application_id"] | ||
updated_association = create_user_application_association(app_id, user_id) | ||
duplicate_return_value = create_user_application_association(app_id, user_id) | ||
assert updated_association is not None | ||
assert duplicate_return_value is None |