diff options
| -rw-r--r-- | src/test/test_api.py | 67 |
1 files changed, 65 insertions, 2 deletions
diff --git a/src/test/test_api.py b/src/test/test_api.py index c72aced..0786363 100644 --- a/src/test/test_api.py +++ b/src/test/test_api.py @@ -1,3 +1,9 @@ +import os +import time +import pytest +import random +import ipaddress + from main import app from fastapi import FastAPI from fastapi import testclient @@ -9,6 +15,63 @@ JWT_TOKEN = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJFUzI1NiJ9.eyJ0eXBlIjoiYWNjZXNzIiwi' + \ JWT_HEADER = {'Authorization': f'Bearer {JWT_TOKEN}'} -def test_get(): +def test_001(): + doc_port = random.randint(1, 65536) + doc_ip = str(ipaddress.IPv4Address(random.randint(1, 0xffffffff))) + doc_asn = str(doc_ip) + '_' + str(doc_port) + + json_data = { + 'ip': doc_ip, + 'port': doc_port, + 'whois_description': 'unittest', + 'asn': doc_asn, + 'asn_country_code': 'SE', + 'ptr': 'unittest.example.com', + 'abuse_mail': 'unittest@example.com', + 'domain': 'sunet.se', + 'timestamp_in_utc': '2021-06-21T14:06UTC', + 'producer_unique_keys': { + 'subject_cn': 'unittest', + 'subject_o': 'unittest', + 'full_name': 'unittest', + 'end_of_general_support': False, + 'cve_2021_21972': 'unittest', + 'cve_2021_21974': 'unittest', + 'cve_2021_21985': 'unittest' + } + } + + response = client.post("/sc/v0/add", headers=JWT_HEADER, json=json_data) + assert(response.status_code == 200) + assert(response.json()['status'] == 'success') + + response = client.get(f"/sc/v0/get?port={doc_port}", headers=JWT_HEADER) + assert(response.status_code == 200) + assert(response.json()['status'] == 'success') + assert(len(response.json()['docs']) == 1) + assert(response.json()['docs'][0]['port'] == doc_port) + + response = client.get(f"/sc/v0/get?asn={doc_asn}", headers=JWT_HEADER) + assert(response.status_code == 200) + assert(response.json()['status'] == 'success') + assert(len(response.json()['docs']) == 1) + assert(response.json()['docs'][0]['asn'] == doc_asn) + + response = client.get(f"/sc/v0/get?ip={doc_ip}", headers=JWT_HEADER) + assert(response.status_code == 200) + assert(response.json()['status'] == 'success') + assert(len(response.json()['docs']) == 1) + assert(response.json()['docs'][0]['ip'] == doc_ip) + + +def test_002(): response = client.get("/sc/v0/get", headers=JWT_HEADER) - print(response.json()) + assert(response.status_code == 200) + + for doc in response.json()['docs']: + doc_id = doc['_id'] + response_doc = client.get(f"/sc/v0/get/{doc_id}", headers=JWT_HEADER) + assert(response_doc.status_code == 200) + assert(response_doc.json()['status'] == 'success') + assert(type(response_doc.json()['docs']) == type(dict())) + assert(response_doc.json()['docs']['domain'] == 'sunet.se') |