chore: update tests

Signed-off-by: Akiff Manji <amanji@petridish.dev>

legal-api/src/legal_api/models/dc_definition.py

@@ -87,11 +87,12 @@ def find_by(cls,
  credential_definition_id: str,
  ) -> DCDefinition:
  """Return the digital credential definition matching the filter."""
- query = db.session.query(DCDefinition). \
- filter(DCDefinition.is_deleted is False). \
- filter(DCDefinition.credential_type == credential_type). \
- filter(DCDefinition.schema_id == schema_id). \
- filter(DCDefinition.credential_definition_id == credential_definition_id)
+ query = (
+ db.session.query(DCDefinition)
+ .filter(DCDefinition.is_deleted == False) # noqa: E712 # pylint: disable=singleton-comparison
+ .filter(DCDefinition.credential_type == credential_type)
+ .filter(DCDefinition.schema_id == schema_id)
+ .filter(DCDefinition.credential_definition_id == credential_definition_id))
  return query.one_or_none()
 
  @classmethod

legal-api/tests/unit/models/test_dc_definition.py

@@ -64,9 +64,9 @@ def test_find_by(session):
  """Assert that the method returns correct value."""
  definition = create_dc_definition()
 
- res = DCDefinition.find_by(DCDefinition.CredentialType.business,
- 'business_schema',
- schema_version='1.0.0'
+ res = DCDefinition.find_by(credential_type=DCDefinition.CredentialType.business,
+ schema_id='test_schema_id',
+ credential_definition_id='test_credential_definition_id',
  )
  assert res
  assert res.id == definition.id
@@ -76,10 +76,10 @@ def create_dc_definition():
  """Create new dc_definition object."""
  definition = DCDefinition(
  credential_type=DCDefinition.CredentialType.business,
- schema_name='business_schema',
+ schema_name='test_business_schema',
  schema_version='1.0.0',
- schema_id='3ENKbWGgUBXXzDHnG11phS:2:business_schema:1.0.0',
- credential_definition_id='3ENKbWGgUBXXzDHnG11phS:3:CL:146949:business_schema'
+ schema_id='test_schema_id',
+ credential_definition_id='test_credential_definition_id'
  )
  definition.save()
  return definition

legal-api/tests/unit/resources/v2/test_business_digital_credentials.py

@@ -34,54 +34,54 @@
 content_type = 'application/json'
 
 
-def test_create_invitation(session, client, jwt): # pylint:disable=unused-argument
+def test_create_invitation(app, session, client, jwt): # pylint:disable=unused-argument
  """Assert create invitation endpoint returns invitation_url."""
  headers = create_header(jwt, [BASIC_USER])
  identifier = 'FM1234567'
  factory_business(identifier)
 
- connection_id = '0d94e18b-3a52-4122-8adf-33e2ccff681f'
+ invitation_id = '0d94e18b-3a52-4122-8adf-33e2ccff681f'
  invitation_url = """http://192.168.65.3:8020?c_i=eyJAdHlwZSI6ICJodHRwczovL2RpZGNvbW0ub3JnL2Nvbm5lY3Rpb
 25zLzEuMC9pbnZpdGF0aW9uIiwgIkBpZCI6ICIyZjU1M2JkZS01YWJlLTRkZDctODIwZi1mNWQ2Mjc1OWQxODgi
 LCAicmVjaXBpZW50S2V5cyI6IFsiMkFHSjVrRDlVYU45OVpSeUFHZVZKNDkxclZhNzZwZGZYdkxXZkFyc2lKWjY
 iXSwgImxhYmVsIjogImZhYmVyLmFnZW50IiwgInNlcnZpY2VFbmRwb2ludCI6ICJodHRwOi8vMTkyLjE2OC42NS4zOjgwMjAifQ=="""
 
  with patch.object(DigitalCredentialsService, 'create_invitation', return_value={
- 'connection_id': connection_id, 'invitation_url': invitation_url}):
+ 'invitation': {'@id': invitation_id}, 'invitation_url': invitation_url}):
+
  rv = client.post(f'/api/v2/businesses/{identifier}/digitalCredentials/invitation',
  headers=headers, content_type=content_type)
  assert rv.status_code == HTTPStatus.OK
  assert rv.json.get('invitationUrl') == invitation_url
 
 
-def test_get_connection_not_found(session, client, jwt): # pylint:disable=unused-argument
- """Assert get connection endpoint returns not found when there is no active connection."""
+def test_get_connections_not_found(session, client, jwt): # pylint:disable=unused-argument
+ """Assert get connections endpoint returns not found when there is no active connection."""
  headers = create_header(jwt, [BASIC_USER])
  identifier = 'FM1234567'
- business = factory_business(identifier)
- create_dc_connection(business)
+ factory_business(identifier)
 
- rv = client.get(f'/api/v2/businesses/{identifier}/digitalCredentials/connection',
+ rv = client.get(f'/api/v2/businesses/{identifier}/digitalCredentials/connections',
  headers=headers, content_type=content_type)
- assert rv.status_code == HTTPStatus.NOT_FOUND
- assert rv.json.get('message') == 'No active connection found.'
+ assert rv.status_code == HTTPStatus.OK
+ assert rv.json.get('connections') == []
 
 
-def test_get_connection(session, client, jwt): # pylint:disable=unused-argument
+def test_get_connections(session, client, jwt): # pylint:disable=unused-argument
  """Assert get connection endpoint returns connection json."""
  headers = create_header(jwt, [BASIC_USER])
  identifier = 'FM1234567'
  business = factory_business(identifier)
 
  connection = create_dc_connection(business, is_active=True)
 
- rv = client.get(f'/api/v2/businesses/{identifier}/digitalCredentials/connection',
+ rv = client.get(f'/api/v2/businesses/{identifier}/digitalCredentials/connections',
  headers=headers, content_type=content_type)
  assert rv.status_code == HTTPStatus.OK
- assert rv.json.get('invitationUrl') == connection.invitation_url
- assert rv.json.get('connectionId') == connection.connection_id
- assert rv.json.get('isActive') == connection.is_active
- assert rv.json.get('connectionState') == connection.connection_state
+ assert rv.json.get('connections')[0].get('invitationUrl') == connection.invitation_url
+ assert rv.json.get('connections')[0].get('connectionId') == connection.connection_id
+ assert rv.json.get('connections')[0].get('isActive') == connection.is_active
+ assert rv.json.get('connections')[0].get('connectionState') == connection.connection_state
 
 
 def test_send_credential(session, client, jwt): # pylint:disable=unused-argument
@@ -94,12 +94,12 @@ def test_send_credential(session, client, jwt): # pylint:disable=unused-argumen
  create_dc_connection(business, is_active=True)
 
  with patch.object(DigitalCredentialsService, 'issue_credential', return_value={
- 'credential_exchange_id': '3fa85f64-5717-4562-b3fc-2c963f66afa6'}):
+ 'cred_ex_id': '3fa85f64-5717-4562-b3fc-2c963f66afa6'}):
  rv = client.post(
  f'/api/v2/businesses/{identifier}/digitalCredentials/{DCDefinition.CredentialType.business.name}',
  headers=headers, content_type=content_type)
  assert rv.status_code == HTTPStatus.OK
- assert rv.json.get('message') == 'Issue Credential is initiated.'
+ assert rv.json.get('message') == 'Credential offer has been sent.'
 
 
 def test_get_issued_credentials(session, client, jwt): # pylint:disable=unused-argument
@@ -130,19 +130,20 @@ def test_webhook_connections_notification(session, client, jwt): # pylint:disab
  connection = create_dc_connection(business)
 
  json_data = {
- 'connection_id': connection.connection_id,
+ 'invitation': {'@id': connection.connection_id},
+ "connection_id": connection.connection_id,
  'state': 'active'
  }
  rv = client.post('/api/v2/digitalCredentials/topic/connections',
  json=json_data,
  headers=headers, content_type=content_type)
  assert rv.status_code == HTTPStatus.OK
 
- rv = client.get(f'/api/v2/businesses/{identifier}/digitalCredentials/connection',
+ rv = client.get(f'/api/v2/businesses/{identifier}/digitalCredentials/connections',
  headers=headers, content_type=content_type)
  assert rv.status_code == HTTPStatus.OK
- assert rv.json.get('isActive') == connection.is_active
- assert rv.json.get('connectionState') == connection.connection_state
+ assert rv.json.get('connections')[0].get('isActive') == connection.is_active
+ assert rv.json.get('connections')[0].get('connectionState') == connection.connection_state
 
 
 def test_webhook_issue_credential_notification(session, client, jwt): # pylint:disable=unused-argument
@@ -154,10 +155,10 @@ def test_webhook_issue_credential_notification(session, client, jwt): # pylint:
  issued_credential = create_dc_issued_credential(business=business)
 
  json_data = {
- 'credential_exchange_id': issued_credential.credential_exchange_id,
- 'state': 'credential_issued'
+ 'cred_ex_id': issued_credential.credential_exchange_id,
+ 'state': 'done'
  }
- rv = client.post('/api/v2/digitalCredentials/topic/issue_credential',
+ rv = client.post('/api/v2/digitalCredentials/topic/issue_credential_v2_0',
  json=json_data,
  headers=headers, content_type=content_type)
  assert rv.status_code == HTTPStatus.OK

legal-api/tests/unit/services/test_digital_credentials.py

@@ -24,10 +24,10 @@
 
 def test_init_app(session, app): # pylint:disable=unused-argument
  """Assert that the init app register schema and credential definition."""
- schema_id = '3ENKbWGgUBXXzDHnG11phS:2:business_schema:1.0.0'
- cred_def_id = '3ENKbWGgUBXXzDHnG11phS:3:CL:146949:business_schema'
- with patch.object(DigitalCredentialsService, '_register_schema', return_value=schema_id):
- with patch.object(DigitalCredentialsService, '_register_credential_definitions', return_value=cred_def_id):
+ schema_id = 'test_schema_id'
+ cred_def_id = 'test_credential_definition_id'
+ with patch.object(DigitalCredentialsService, '_fetch_schema', return_value=schema_id):
+ with patch.object(DigitalCredentialsService, '_fetch_credential_definition', return_value=cred_def_id):
  digital_credentials.init_app(app)
  definition = DCDefinition.find_by_credential_type(DCDefinition.CredentialType.business)
  assert definition.schema_id == schema_id

legal-api/wsgi.py

@@ -19,8 +19,8 @@
 from legal_api.extensions import socketio
 
 # Openshift s2i expects a lower case name of application
-application = create_app() # pylint: disable=invalid-name
+application = create_app()  # pylint: disable=invalid-name
 
 if __name__ == "__main__":
  server_port = os.environ.get('PORT', '8080')
- socketio.run(application, port=server_port, host='0.0.0.0')
+ socketio.run(application, debug=False, port=server_port, host='0.0.0.0')