kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Minor refactors:

Browse Source 
- Renames s_assemble to s_brief
-  Link s_local to s_brief
This commit is contained in:
Louis Holbrook
2021-03-04 16:47:13 +00:00
committed by Philip Wafula
parent 21c9d95c4b
commit 1e7fff0133
93 changed files with 2264 additions and 796 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
apps/cic-ussd/tests/cic_ussd/metadata/test_metadata.py Normal file
View File

@@ -0,0 +1,80 @@
# standard imports
import json
# third-party imports
import pytest
import requests
import requests_mock
# local imports
from cic_ussd.error import UnsupportedMethodError
from cic_ussd.metadata import blockchain_address_to_metadata_pointer, make_request
def test_make_request(define_metadata_pointer_url, mock_meta_get_response, mock_meta_post_response, person_metadata):
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'GET',
define_metadata_pointer_url,
status_code=200,
reason='OK',
content=json.dumps(mock_meta_get_response).encode('utf-8')
)
response = make_request(method='GET', url=define_metadata_pointer_url)
assert response.content == requests.get(define_metadata_pointer_url).content
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'POST',
define_metadata_pointer_url,
status_code=201,
reason='CREATED',
content=json.dumps(mock_meta_post_response).encode('utf-8')
)
response = make_request(
method='POST',
url=define_metadata_pointer_url,
data=json.dumps(person_metadata).encode('utf-8'),
headers={
'X-CIC-AUTOMERGE': 'server',
'Content-Type': 'application/json'
}
)
assert response.content == requests.post(define_metadata_pointer_url).content
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'PUT',
define_metadata_pointer_url,
status_code=200,
reason='OK'
)
response = make_request(
method='PUT',
url=define_metadata_pointer_url,
data=json.dumps(person_metadata).encode('utf-8'),
headers={
'X-CIC-AUTOMERGE': 'server',
'Content-Type': 'application/json'
}
)
assert response.content == requests.put(define_metadata_pointer_url).content
with pytest.raises(UnsupportedMethodError) as error:
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'DELETE',
define_metadata_pointer_url,
status_code=200,
reason='OK'
)
make_request(
method='DELETE',
url=define_metadata_pointer_url
)
assert str(error.value) == 'Unsupported method: DELETE'
def test_blockchain_address_to_metadata_pointer(create_activated_user):
blockchain_address = create_activated_user.blockchain_address
assert type(blockchain_address_to_metadata_pointer(blockchain_address)) == bytes

34
apps/cic-ussd/tests/cic_ussd/metadata/test_signer.py Normal file
View File

@@ -0,0 +1,34 @@
# standard imports
import shutil
# third-party imports
# local imports
from cic_ussd.metadata.signer import Signer
def test_client(load_config, setup_metadata_signer, person_metadata):
signer = Signer()
# get gpg used
digest = 'a4337bc45a8fc544c03f52dc550cd6e1e87021bc896588bd79e901e2'
person_metadata['digest'] = digest
gpg = signer.gpg
# check that key data was loaded
assert signer.key_data is not None
# check that correct operational key is returned
gpg.import_keys(key_data=signer.key_data)
gpg_keys = gpg.list_keys()
assert signer.get_operational_key() == gpg_keys[0]
# check that correct signature is returned
key_id = signer.get_operational_key().get('keyid')
signature = gpg.sign(message=digest, passphrase=load_config.get('KEYS_PASSPHRASE'), keyid=key_id)
assert str(signature) == signer.sign_digest(data=person_metadata)
# remove tmp gpg file
shutil.rmtree(Signer.gpg_path)

123
apps/cic-ussd/tests/cic_ussd/metadata/test_user_metadata.py Normal file
View File

@@ -0,0 +1,123 @@
# standard imports
import json
# third-party imports
import pytest
import requests_mock
from cic_types.models.person import generate_metadata_pointer
# local imports
from cic_ussd.metadata import blockchain_address_to_metadata_pointer
from cic_ussd.metadata.signer import Signer
from cic_ussd.metadata.user import UserMetadata
from cic_ussd.redis import get_cached_data
def test_user_metadata(create_activated_user, define_metadata_pointer_url, load_config):
UserMetadata.base_url = load_config.get('CIC_META_URL')
identifier = blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address)
user_metadata_client = UserMetadata(identifier=identifier)
assert user_metadata_client.url == define_metadata_pointer_url
def test_create_user_metadata(caplog,
create_activated_user,
define_metadata_pointer_url,
load_config,
mock_meta_post_response,
person_metadata):
identifier = blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address)
user_metadata_client = UserMetadata(identifier=identifier)
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'POST',
define_metadata_pointer_url,
status_code=201,
reason='CREATED',
content=json.dumps(mock_meta_post_response).encode('utf-8')
)
user_metadata_client.create(data=person_metadata)
assert 'Get signed material response status: 201' in caplog.text
with pytest.raises(RuntimeError) as error:
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'POST',
define_metadata_pointer_url,
status_code=400,
reason='BAD REQUEST'
)
user_metadata_client.create(data=person_metadata)
assert str(error.value) == f'400 Client Error: BAD REQUEST for url: {define_metadata_pointer_url}'
def test_edit_user_metadata(caplog,
create_activated_user,
define_metadata_pointer_url,
load_config,
person_metadata,
setup_metadata_signer):
Signer.gpg_passphrase = load_config.get('KEYS_PASSPHRASE')
identifier = blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address)
user_metadata_client = UserMetadata(identifier=identifier)
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'PUT',
define_metadata_pointer_url,
status_code=200,
reason='OK'
)
user_metadata_client.edit(data=person_metadata, engine='pgp')
assert 'Signed content submission status: 200' in caplog.text
with pytest.raises(RuntimeError) as error:
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'PUT',
define_metadata_pointer_url,
status_code=400,
reason='BAD REQUEST'
)
user_metadata_client.edit(data=person_metadata, engine='pgp')
assert str(error.value) == f'400 Client Error: BAD REQUEST for url: {define_metadata_pointer_url}'
def test_get_user_metadata(caplog,
create_activated_user,
define_metadata_pointer_url,
init_redis_cache,
load_config,
person_metadata,
setup_metadata_signer):
identifier = blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address)
user_metadata_client = UserMetadata(identifier=identifier)
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'GET',
define_metadata_pointer_url,
status_code=200,
content=json.dumps(person_metadata).encode('utf-8'),
reason='OK'
)
user_metadata_client.query()
assert 'Get latest data status: 200' in caplog.text
key = generate_metadata_pointer(
identifier=identifier,
cic_type='cic.person'
)
cached_user_metadata = get_cached_data(key=key)
assert cached_user_metadata
with pytest.raises(RuntimeError) as error:
with requests_mock.Mocker(real_http=False) as request_mocker:
request_mocker.register_uri(
'GET',
define_metadata_pointer_url,
status_code=404,
reason='NOT FOUND'
)
user_metadata_client.query()
assert 'The data is not available and might need to be added.' in caplog.text
assert str(error.value) == f'400 Client Error: NOT FOUND for url: {define_metadata_pointer_url}'