Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found
Select Git revision

Target

Select target project
  • fair-data-austria-db-repository/fda-services
1 result
Select Git revision
Show changes
Show whitespace changes
Inline Side-by-side
Showing
with 1465 additions and 425 deletions
lib/python/tests/test_unit_analyse.py → lib/python/tests/test_unit_messages.py
View file @ 841d4894
import unittest
import requests_mock
from dbrepo.RestClient import RestClient
from dbrepo.api.dto import Message
from dbrepo.api.dto import KeyAnalysis
class ImageUnitTest(unittest.TestCase):
class AnalyseUnitTest(unittest.TestCase):
def test_get_message_empty_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/message', json=[])
# test
response = RestClient().get_messages()
self.assertEqual([], response)
def test_analyse_keys_succeeds(self):
def test_get_images_succeeds(self):
with requests_mock.Mocker() as mock:
exp = KeyAnalysis(keys={'id': 0, 'firstname': 1, 'lastname': 2})
exp = [Message(id="a456d7f0-9d42-48a8-bf5b-4ead85279e0e", type="info")]
# mock
mock.get('/api/analyse/keys', json=exp.model_dump(), status_code=202)
mock.get('/api/message', json=[exp[0].model_dump()])
# test
response = RestClient().analyse_keys(file_path='f705a7bd0cb2d5e37ab2b425036810a2', separator=',',
upload=False)
response = RestClient().get_messages()
self.assertEqual(exp, response)
......
lib/python/tests/test_unit_ontology.py 0 → 100644
View file @ 841d4894
import unittest
import requests_mock
from dbrepo.RestClient import RestClient
from dbrepo.api.dto import OntologyBrief
class OntologyUnitTest(unittest.TestCase):
def test_get_ontologies_empty_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/ontology', json=[])
# test
response = RestClient().get_ontologies()
self.assertEqual([], response)
def test_get_ontologies_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [OntologyBrief(id="ec6ee082-b6b2-4d65-b931-2f1735f25759",
uri="http://www.ontology-of-units-of-measure.org/resource/om-2/",
prefix="om",
sparql=False,
rdf=True,
uri_pattern="http://www.ontology-of-units-of-measure.org/resource/om-2/.*")]
# mock
mock.get('/api/ontology', json=[exp[0].model_dump()])
# test
response = RestClient().get_ontologies()
self.assertEqual(exp, response)
if __name__ == "__main__":
unittest.main()
lib/python/tests/test_unit_query.py
View file @ 841d4894
import datetime
import json
import unittest
import requests_mock
import datetime
from dbrepo.RestClient import RestClient
from pandas import DataFrame
from dbrepo.api.dto import Query, User, UserAttributes, QueryType, UserBrief
from dbrepo.api.exceptions import MalformedError, NotExistsError, ForbiddenError
from dbrepo.RestClient import RestClient
from dbrepo.api.dto import Query, QueryType, UserBrief, QueryDefinition, FilterDefinition, FilterType, Database, \
ContainerBrief, ImageBrief, Image, Table, Constraints, PrimaryKey, TableBrief, ColumnBrief, ColumnType, \
Column, Operator
from dbrepo.api.exceptions import MalformedError, NotExistsError, ForbiddenError, QueryStoreError, FormatNotAvailable, \
ServiceError, ResponseCodeError, AuthenticationError
class QueryUnitTest(unittest.TestCase):
database = Database(
id="6bd39359-b154-456d-b9c2-caa516a45732",
name='test',
owner=UserBrief(id='abdbf897-e599-4e5a-a3f0-7529884ea011', username='mweise'),
contact=UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise'),
exchange_name='dbrepo',
internal_name='test_abcd',
is_public=True,
is_schema_public=True,
tables=[Table(id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
name="Some Table",
description="Test Table",
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
internal_name="some_table",
owner=UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise'),
is_versioned=True,
queue_name='test',
routing_key='dbrepo.test_database_1234.test',
is_public=True,
is_schema_public=True,
constraints=Constraints(uniques=[],
foreign_keys=[],
checks=[],
primary_key=[PrimaryKey(id="1516310f-ecb5-4614-abe2-3b96114e1484",
table=TableBrief(
id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
name='Other',
internal_name='other',
description=None,
is_versioned=True,
is_public=True,
is_schema_public=True,
owned_by='8638c043-5145-4be8-a3e4-4b79991b0a16'),
column=ColumnBrief(
id="31a533b6-8ddf-43d6-ac6a-b9da597cb976",
table_id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
name='id',
alias=None,
internal_name='id',
type=ColumnType.BIGINT))]),
columns=[Column(id="31a533b6-8ddf-43d6-ac6a-b9da597cb976",
name="ID",
ord=0,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
internal_name="id",
type=ColumnType.BIGINT,
is_null_allowed=False),
Column(id="85de93a8-834c-4cf4-9d34-f80ebd97e606",
name="Username",
ord=1,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
internal_name="username",
type=ColumnType.VARCHAR,
is_null_allowed=False)
])],
container=ContainerBrief(id="44d811a8-4019-46ba-bd57-ea10a2eb0c74",
name='MariaDB Galera 11.1.3',
internal_name='mariadb',
image=ImageBrief(id="b104648b-54d2-4d72-9834-8e0e6d428b39",
name='mariadb',
version='11.2.2',
default=True)))
image = Image(id="b104648b-54d2-4d72-9834-8e0e6d428b39",
name='mariadb',
version='11.2.2',
default=True,
data_types=[],
operators=[Operator(id="6a96bd99-be3d-4d56-8c38-b14bdfead634",
display_name="IN",
value="IN",
documentation="https://mariadb.com/kb/en/in/")])
def test_create_subset_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [{'id': 1, 'username': 'foo'}, {'id': 2, 'username': 'bar'}]
df = DataFrame.from_records(json.dumps(exp))
# mock
mock.post('/api/database/1/subset', json=json.dumps(exp), headers={'X-Id': '1'}, status_code=201)
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', json=json.dumps(exp),
headers={'X-Id': '1'}, status_code=201)
# test
client = RestClient(username="a", password="b")
response = client.create_subset(database_id=1, page=0, size=10,
query="SELECT id, username FROM some_table WHERE id IN (1,2)")
response = client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732", page=0, size=10,
timestamp=datetime.datetime(2024, 1, 1, 0, 0, 0, 0, datetime.timezone.utc),
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
self.assertTrue(DataFrame.equals(df, response))
def test_create_subset_malformed_fails(self):
def test_create_subset_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/subset', status_code=400)
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', status_code=400)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_subset(database_id=1,
query="SELECT id, username FROM some_table WHERE id IN (1,2)")
client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
except MalformedError:
pass
def test_create_subset_not_allowed_fails(self):
def test_create_subset_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/subset', status_code=403)
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', status_code=403)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_subset(database_id=1,
query="SELECT id, username FROM some_table WHERE id IN (1,2)")
client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
except ForbiddenError:
pass
def test_create_subset_not_found_fails(self):
def test_create_subset_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/subset', status_code=404)
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', status_code=404)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_subset(database_id=1,
query="SELECT id, username FROM some_table WHERE id IN (1,2)")
client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
except NotExistsError:
pass
def test_create_subset_not_auth_succeeds(self):
def test_create_subset_417_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', status_code=417)
# test
try:
client = RestClient(username="a", password="b")
client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
except QueryStoreError:
pass
def test_create_subset_501_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', status_code=501)
# test
try:
client = RestClient(username="a", password="b")
client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
except FormatNotAvailable:
pass
def test_create_subset_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', status_code=503)
# test
try:
client = RestClient(username="a", password="b")
client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
except ServiceError:
pass
def test_create_subset_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', status_code=200)
# test
try:
client = RestClient(username="a", password="b")
client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
except ResponseCodeError:
pass
def test_create_subset_anonymous_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [{'id': 1, 'username': 'foo'}, {'id': 2, 'username': 'bar'}]
df = DataFrame.from_records(json.dumps(exp))
# mock
mock.post('/api/database/1/subset', json=json.dumps(exp), headers={'X-Id': '1'}, status_code=201)
mock.get(f'/api/image/{self.image.id}', json=self.image.model_dump(),
status_code=200)
mock.get(f'/api/database/{self.database.id}', json=self.database.model_dump(),
status_code=200)
mock.post(f'/api/database/{self.database.id}/subset', json=json.dumps(exp),
headers={'X-Id': '1'}, status_code=201)
# test
client = RestClient()
response = client.create_subset(database_id=1, page=0, size=10,
query="SELECT id, username FROM some_table WHERE id IN (1,2)")
response = client.create_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732", page=0, size=10,
query=QueryDefinition(table="some_table",
columns=["id", "username"],
filter=[FilterDefinition(type=FilterType.WHERE,
column="id",
operator="IN",
value="(1,2)")]))
self.assertTrue(DataFrame.equals(df, response))
def test_find_query_succeeds(self):
def test_get_subset_succeeds(self):
with requests_mock.Mocker() as mock:
exp = Query(id=6,
exp = Query(id="e1df2bb8-1f12-494a-ade5-2c4aecdab939",
owner=UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise'),
execution=datetime.datetime(2024, 1, 1, 0, 0, 0, 0, datetime.timezone.utc),
query='SELECT id, username FROM some_table WHERE id IN (1,2)',
query_normalized='SELECT id, username FROM some_table WHERE id IN (1,2)',
type=QueryType.QUERY,
database_id=1,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query_hash='da5ff66c4a57683171e2ffcec25298ee684680d1e03633cd286f9067d6924ad8',
result_hash='464740ba612225913bb15b26f13377707949b55e65288e89c3f8b4c6469aecb4',
is_persisted=False,
result_number=None,
identifiers=[])
# mock
mock.get('/api/database/1/subset/6', json=exp.model_dump())
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
json=exp.model_dump())
# test
response = RestClient().get_subset(database_id=1, subset_id=6)
response = RestClient().get_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
self.assertEqual(exp, response)
def test_find_query_not_allowed_fails(self):
def test_get_subset_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/subset/6', status_code=403)
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=403)
# test
try:
response = RestClient().get_subset(database_id=1, subset_id=6)
RestClient().get_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ForbiddenError:
pass
def test_find_query_not_found_fails(self):
def test_get_subset_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/subset/6', status_code=404)
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=404)
# test
try:
response = RestClient().get_subset(database_id=1, subset_id=6)
RestClient().get_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except NotExistsError:
pass
def test_get_subset_406_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=406)
# test
try:
RestClient().get_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except FormatNotAvailable:
pass
def test_get_subset_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=503)
# test
try:
RestClient().get_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ServiceError:
pass
def test_get_subset_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=202)
# test
try:
RestClient().get_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ResponseCodeError:
pass
def test_get_queries_empty_succeeds(self):
with requests_mock.Mocker() as mock:
exp = []
# mock
mock.get('/api/database/1/subset', json=[])
mock.get(f'/api/database/{self.database.id}/subset', json=[])
# test
response = RestClient().get_queries(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
self.assertEqual(exp, response)
def test_update_subset_succeeds(self):
with requests_mock.Mocker() as mock:
exp = Query(id="e1df2bb8-1f12-494a-ade5-2c4aecdab939",
owner=UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise'),
execution=datetime.datetime(2024, 1, 1, 0, 0, 0, 0, datetime.timezone.utc),
query='SELECT id, username FROM some_table WHERE id IN (1,2)',
query_normalized='SELECT id, username FROM some_table WHERE id IN (1,2)',
type=QueryType.QUERY,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query_hash='da5ff66c4a57683171e2ffcec25298ee684680d1e03633cd286f9067d6924ad8',
result_hash='464740ba612225913bb15b26f13377707949b55e65288e89c3f8b4c6469aecb4',
is_persisted=True,
result_number=None,
identifiers=[])
# mock
mock.put('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
json=exp.model_dump(),
status_code=202)
# test
response = RestClient().get_queries(database_id=1)
response = RestClient(username='foo', password='bar').update_subset(
database_id="6bd39359-b154-456d-b9c2-caa516a45732", subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939",
persist=True)
self.assertEqual(exp, response)
def test_update_subset_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=400)
# test
try:
RestClient(username='foo', password='bar').update_subset(
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939", persist=True)
except MalformedError:
pass
def test_update_subset_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=403)
# test
try:
RestClient(username='foo', password='bar').update_subset(
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939", persist=True)
except ForbiddenError:
pass
def test_update_subset_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=404)
# test
try:
RestClient(username='foo', password='bar').update_subset(
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939", persist=True)
except NotExistsError:
pass
def test_update_subset_417_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=417)
# test
try:
RestClient(username='foo', password='bar').update_subset(
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939", persist=True)
except QueryStoreError:
pass
def test_update_subset_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=503)
# test
try:
RestClient(username='foo', password='bar').update_subset(
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939", persist=True)
except ServiceError:
pass
def test_update_subset_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939',
status_code=200)
# test
try:
RestClient(username='foo', password='bar').update_subset(
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939",
persist=True)
except ResponseCodeError:
pass
def test_update_subset_anonymous_fails(self):
# test
try:
RestClient().update_subset(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939",
persist=True)
except AuthenticationError:
pass
def test_get_queries_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [Query(id=6,
exp = [Query(id="e1df2bb8-1f12-494a-ade5-2c4aecdab939",
owner=UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise'),
execution=datetime.datetime(2024, 1, 1, 0, 0, 0, 0, datetime.timezone.utc),
query='SELECT id, username FROM some_table WHERE id IN (1,2)',
query_normalized='SELECT id, username FROM some_table WHERE id IN (1,2)',
type=QueryType.QUERY,
database_id=1,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
query_hash='da5ff66c4a57683171e2ffcec25298ee684680d1e03633cd286f9067d6924ad8',
result_hash='464740ba612225913bb15b26f13377707949b55e65288e89c3f8b4c6469aecb4',
is_persisted=False,
result_number=None,
identifiers=[])]
# mock
mock.get('/api/database/1/subset', json=[exp[0].model_dump()])
mock.get(f'/api/database/{self.database.id}/subset', json=[exp[0].model_dump()])
# test
response = RestClient().get_queries(database_id=1)
response = RestClient().get_queries(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
self.assertEqual(exp, response)
def test_get_queries_not_allowed_fails(self):
def test_get_queries_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/subset', status_code=403)
mock.get(f'/api/database/{self.database.id}/subset', status_code=403)
# test
try:
response = RestClient().get_queries(database_id=1)
RestClient().get_queries(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
except ForbiddenError:
pass
def test_get_queries_not_found_fails(self):
def test_get_queries_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/subset', status_code=404)
mock.get(f'/api/database/{self.database.id}/subset', status_code=404)
# test
try:
response = RestClient().get_queries(database_id=1)
RestClient().get_queries(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
except NotExistsError:
pass
def test_get_queries_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(f'/api/database/{self.database.id}/subset', status_code=503)
# test
try:
RestClient().get_queries(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
except ServiceError:
pass
def test_get_queries_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(f'/api/database/{self.database.id}/subset', status_code=202)
# test
try:
RestClient().get_queries(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
except ResponseCodeError:
pass
def test_get_subset_data_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [{'id': 1, 'username': 'foo'}, {'id': 2, 'username': 'bar'}]
df = DataFrame.from_records(json.dumps(exp))
# mock
mock.get('/api/database/1/subset/6/data', json=json.dumps(exp))
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
json=json.dumps(exp))
# test
response = RestClient().get_subset_data(database_id=1, subset_id=6)
response = RestClient().get_subset_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
self.assertTrue(DataFrame.equals(df, response))
def test_get_subset_data_dataframe_succeeds(self):
......@@ -178,61 +564,157 @@ class QueryUnitTest(unittest.TestCase):
exp = [{'id': 1, 'username': 'foo'}, {'id': 2, 'username': 'bar'}]
df = DataFrame.from_records(json.dumps(exp))
# mock
mock.get('/api/database/1/subset/6/data', json=json.dumps(exp))
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
json=json.dumps(exp))
# test
response = RestClient().get_subset_data(database_id=1, subset_id=6)
response = RestClient().get_subset_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
self.assertEqual(df.shape, response.shape)
self.assertTrue(DataFrame.equals(df, response))
def test_get_subset_data_not_allowed_fails(self):
def test_get_subset_data_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=400)
# test
try:
RestClient().get_subset_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except MalformedError:
pass
def test_get_subset_data_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/subset/6/data', status_code=403)
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=403)
# test
try:
response = RestClient().get_subset_data(database_id=1, subset_id=6)
RestClient().get_subset_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ForbiddenError:
pass
def test_get_subset_data_not_found_fails(self):
def test_get_subset_data_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/subset/6/data', status_code=404)
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=404)
# test
try:
response = RestClient().get_subset_data(database_id=1, subset_id=6)
RestClient().get_subset_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except NotExistsError:
pass
def test_get_subset_data_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=503)
# test
try:
RestClient().get_subset_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ServiceError:
pass
def test_get_subset_data_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=202)
# test
try:
RestClient().get_subset_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ResponseCodeError:
pass
def test_get_subset_data_count_succeeds(self):
with requests_mock.Mocker() as mock:
exp = 2
# mock
mock.head('/api/database/1/subset/6/data', headers={'X-Count': str(exp)})
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
headers={'X-Count': str(exp)})
# test
response = RestClient().get_subset_data_count(database_id=1, subset_id=6)
response = RestClient().get_subset_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
self.assertEqual(exp, response)
def test_get_subset_data_count_not_allowed_fails(self):
def test_get_subset_data_count_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head('/api/database/1/subset/6/data', status_code=403)
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=400)
# test
try:
response = RestClient().get_subset_data_count(database_id=1, subset_id=6)
RestClient().get_subset_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except MalformedError:
pass
def test_get_subset_data_count_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=403)
# test
try:
RestClient().get_subset_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ForbiddenError:
pass
def test_get_subset_data_count_not_found_fails(self):
def test_get_subset_data_count_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head('/api/database/1/subset/6/data', status_code=404)
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=404)
# test
try:
response = RestClient().get_subset_data_count(database_id=1, subset_id=6)
RestClient().get_subset_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except NotExistsError:
pass
def test_get_subset_data_count_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=503)
# test
try:
RestClient().get_subset_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ServiceError:
pass
def test_get_subset_data_count_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/subset/e1df2bb8-1f12-494a-ade5-2c4aecdab939/data',
status_code=202)
# test
try:
RestClient().get_subset_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
subset_id="e1df2bb8-1f12-494a-ade5-2c4aecdab939")
except ResponseCodeError:
pass
if __name__ == "__main__":
unittest.main()
lib/python/tests/test_unit_rest_client.py
View file @ 841d4894
......@@ -4,96 +4,34 @@ import unittest
import requests_mock
from dbrepo.RestClient import RestClient
from dbrepo.api.dto import JwtAuth
from dbrepo.api.exceptions import MalformedError, ServiceConnectionError, ServiceError, ForbiddenError, \
AuthenticationError, ResponseCodeError
class RestClientUnitTest(unittest.TestCase):
def test_constructor_succeeds(self):
with requests_mock.Mocker() as mock:
# test
os.environ['REST_API_SECURE'] = 'True'
response = RestClient()
self.assertTrue(response.secure)
def test_get_jwt_auth_empty_succeeds(self):
with requests_mock.Mocker() as mock:
exp = JwtAuth(access_token='ey123',
refresh_token='ey456',
id_token='ey789',
expires_in=3600,
scope='scope',
token_type='Bearer',
not_before_policy=0,
session_state='session_state',
refresh_expires_in=7200)
# mock
mock.post('/api/user/token', json=exp.model_dump(), status_code=202)
# test
response = RestClient().get_jwt_auth()
self.assertEqual(exp, response)
def test_get_jwt_auth_400_fails(self):
def test_constructor_token_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/user/token', status_code=400)
mock.get('/api/user', json=[])
# test
try:
response = RestClient().get_jwt_auth()
except MalformedError:
pass
client = RestClient(password='bar')
client.get_users()
self.assertEqual('bar', client.password)
def test_get_jwt_auth_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/user/token', status_code=403)
# test
try:
response = RestClient().get_jwt_auth()
except ForbiddenError:
pass
def test_get_jwt_auth_428_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/user/token', status_code=428)
def test_whoami_anonymous_succeeds(self):
# test
try:
response = RestClient().get_jwt_auth()
except AuthenticationError:
pass
response = RestClient().whoami()
self.assertIsNone(response)
def test_get_jwt_auth_502_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/user/token', status_code=502)
# test
try:
response = RestClient().get_jwt_auth()
except ServiceConnectionError:
pass
def test_get_jwt_auth_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/user/token', status_code=503)
# test
try:
response = RestClient().get_jwt_auth()
except ServiceError:
pass
def test_get_jwt_auth_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/user/token', status_code=418)
def test_whoami_succeeds(self):
# test
try:
response = RestClient().get_jwt_auth()
except ResponseCodeError:
pass
response = RestClient(username="foobar").whoami()
self.assertEqual("foobar", response)
if __name__ == "__main__":
......
lib/python/tests/test_unit_table.py
View file @ 841d4894
import datetime
import json
import unittest
......@@ -6,16 +7,16 @@ from pandas import DataFrame
from dbrepo.RestClient import RestClient
from dbrepo.api.dto import Table, CreateTableConstraints, Column, Constraints, ColumnType, ConceptBrief, UnitBrief, \
TableStatistics, ColumnStatistic, PrimaryKey, ColumnBrief, TableBrief, UserBrief
TableStatistics, ColumnStatistic, PrimaryKey, ColumnBrief, TableBrief, UserBrief, History, HistoryEventType
from dbrepo.api.exceptions import MalformedError, ForbiddenError, NotExistsError, NameExistsError, \
AuthenticationError, ExternalSystemError
AuthenticationError, ExternalSystemError, ServiceError, ServiceConnectionError, ResponseCodeError
class TableUnitTest(unittest.TestCase):
def test_create_table_succeeds(self):
exp = TableBrief(id=2,
database_id=1,
exp = TableBrief(id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
name="Test",
description="Test Table",
internal_name="test",
......@@ -25,73 +26,122 @@ class TableUnitTest(unittest.TestCase):
is_schema_public=True)
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table', json=exp.model_dump(), status_code=201)
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', json=exp.model_dump(),
status_code=201)
# test
client = RestClient(username="a", password="b")
response = client.create_table(database_id=1, name="Test", description="Test Table", columns=[],
response = client.create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732", name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
self.assertEqual(exp, response)
def test_create_table_malformed_fails(self):
def test_create_table_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table', status_code=400)
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=400)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_table(database_id=1, name="Test", description="Test Table", columns=[],
response = client.create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732", name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
except MalformedError:
pass
def test_create_table_not_allowed_fails(self):
def test_create_table_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table', status_code=403)
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=403)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_table(database_id=1, name="Test", description="Test Table", columns=[],
RestClient(username="a", password="b").create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
except ForbiddenError:
pass
def test_create_table_not_allowed_fails(self):
def test_create_table_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table', status_code=404)
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=404)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_table(database_id=1, name="Test", description="Test Table", columns=[],
response = client.create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732", name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
except NotExistsError:
pass
def test_create_table_name_exists_fails(self):
def test_create_table_409_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table', status_code=409)
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=409)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_table(database_id=1, name="Test", description="Test Table", columns=[],
response = client.create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732", name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
except NameExistsError:
pass
def test_create_table_not_auth_fails(self):
def test_create_table_502_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=502)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732", name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
except ServiceConnectionError:
pass
def test_create_table_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=503)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732", name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
except ServiceError:
pass
def test_create_table_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=200)
# test
try:
client = RestClient(username="a", password="b")
response = client.create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732", name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
except ResponseCodeError:
pass
def test_create_table_anonymous_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table', status_code=409)
mock.post('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=409)
# test
try:
response = RestClient().create_table(database_id=1, name="Test", description="Test Table", columns=[],
response = RestClient().create_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732", name="Test",
description="Test Table", columns=[],
is_public=True, is_schema_public=True,
constraints=CreateTableConstraints())
except AuthenticationError:
......@@ -100,34 +150,64 @@ class TableUnitTest(unittest.TestCase):
def test_get_tables_empty_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/table', json=[])
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', json=[])
# test
response = RestClient().get_tables(database_id=1)
response = RestClient().get_tables(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
self.assertEqual([], response)
def test_get_tables_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [TableBrief(id=2,
exp = [TableBrief(id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
name="Test",
description="Test Table",
database_id=1,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
internal_name="test",
is_public=True,
is_schema_public=True,
owned_by='8638c043-5145-4be8-a3e4-4b79991b0a16',
is_versioned=True)]
# mock
mock.get('/api/database/1/table', json=[exp[0].model_dump()])
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', json=[exp[0].model_dump()])
# test
response = RestClient().get_tables(database_id=1)
response = RestClient().get_tables(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
self.assertEqual(exp, response)
def test_get_tables_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=403)
# test
try:
RestClient().get_tables(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
except ForbiddenError:
pass
def test_get_tables_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=404)
# test
try:
RestClient().get_tables(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
except NotExistsError:
pass
def test_get_tables_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table', status_code=202)
# test
try:
RestClient().get_tables(database_id="6bd39359-b154-456d-b9c2-caa516a45732")
except ResponseCodeError:
pass
def test_get_table_succeeds(self):
with requests_mock.Mocker() as mock:
exp = Table(id=2,
exp = Table(id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
name="Test",
description="Test Table",
database_id=1,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
internal_name="test",
owner=UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise'),
is_versioned=True,
......@@ -138,8 +218,10 @@ class TableUnitTest(unittest.TestCase):
constraints=Constraints(uniques=[],
foreign_keys=[],
checks=[],
primary_key=[PrimaryKey(id=1,
table=TableBrief(id=2, database_id=1,
primary_key=[PrimaryKey(id="1516310f-ecb5-4614-abe2-3b96114e1484",
table=TableBrief(
id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
name='Other',
internal_name='other',
description=None,
......@@ -147,97 +229,258 @@ class TableUnitTest(unittest.TestCase):
is_public=True,
is_schema_public=True,
owned_by='8638c043-5145-4be8-a3e4-4b79991b0a16'),
column=ColumnBrief(id=1, table_id=2,
database_id=1,
column=ColumnBrief(
id="31a533b6-8ddf-43d6-ac6a-b9da597cb976",
table_id="",
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
name='id',
alias=None,
internal_name='id',
type=ColumnType.BIGINT))]),
columns=[Column(id=1,
columns=[Column(id="31a533b6-8ddf-43d6-ac6a-b9da597cb976",
name="ID",
ord=0,
database_id=1,
table_id=2,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
internal_name="id",
auto_generated=True,
is_primary_key=True,
type=ColumnType.BIGINT,
is_public=True,
is_null_allowed=False)])
# mock
mock.get('/api/database/1/table/2', json=exp.model_dump())
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
json=exp.model_dump())
# test
response = RestClient().get_table(database_id=1, table_id=2)
response = RestClient().get_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
self.assertEqual(exp, response)
def test_get_table_not_allowed_fails(self):
def test_get_table_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/table/2', status_code=403)
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=403)
# test
try:
response = RestClient().get_table(database_id=1, table_id=2)
response = RestClient().get_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ForbiddenError:
pass
def test_get_table_not_found_fails(self):
def test_get_table_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/table/2', status_code=404)
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=404)
# test
try:
response = RestClient().get_table(database_id=1, table_id=2)
response = RestClient().get_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except NotExistsError:
pass
def test_get_table_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=202)
# test
try:
response = RestClient().get_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ResponseCodeError:
pass
def test_delete_table_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/2', status_code=202)
mock.delete('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=202)
# test
client = RestClient(username="a", password="b")
client.delete_table(database_id=1, table_id=2)
client.delete_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
def test_delete_table_not_allowed_fails(self):
def test_delete_table_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/2', status_code=403)
mock.delete('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=400)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table(database_id=1, table_id=2)
client.delete_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except MalformedError:
pass
def test_delete_table_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=403)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ForbiddenError:
pass
def test_delete_table_not_found_fails(self):
def test_delete_table_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/2', status_code=404)
mock.delete('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=404)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table(database_id=1, table_id=2)
client.delete_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except NotExistsError:
pass
def test_delete_table_not_auth_fails(self):
def test_delete_table_502_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=502)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ServiceConnectionError:
pass
def test_delete_table_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/2', status_code=404)
mock.delete('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=503)
# test
try:
RestClient().delete_table(database_id=1, table_id=2)
client = RestClient(username="a", password="b")
client.delete_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ServiceError:
pass
def test_delete_table_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=200)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ResponseCodeError:
pass
def test_delete_table_anonymous_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692',
status_code=404)
# test
try:
RestClient().delete_table(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except AuthenticationError:
pass
def test_get_table_history_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [History(event=HistoryEventType.INSERT,
total=2,
timestamp=datetime.datetime(2024, 1, 1, 0, 0, 0, 0, datetime.timezone.utc))]
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/history?size=100',
json=[exp[0].model_dump()])
# test
response = RestClient().get_table_history(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
self.assertEqual(exp, response)
def test_get_table_history_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/history?size=100',
status_code=400)
# test
try:
RestClient().get_table_history(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except MalformedError:
pass
def test_get_table_history_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/history?size=100',
status_code=403)
# test
try:
RestClient().get_table_history(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ForbiddenError:
pass
def test_get_table_history_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/history?size=100',
status_code=404)
# test
try:
RestClient().get_table_history(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except NotExistsError:
pass
def test_get_table_history_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/history?size=100',
status_code=503)
# test
try:
RestClient().get_table_history(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ServiceError:
pass
def test_get_table_history_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/history?size=100',
status_code=202)
# test
try:
RestClient().get_table_history(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ResponseCodeError:
pass
def test_get_table_data_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [{'id': 1, 'username': 'foo'}, {'id': 2, 'username': 'bar'}]
df = DataFrame.from_records(json.dumps(exp))
# mock
mock.get('/api/database/1/table/9/data', json=json.dumps(exp))
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
json=json.dumps(exp))
# test
response = RestClient().get_table_data(database_id=1, table_id=9)
response = RestClient().get_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
timestamp=datetime.datetime(2024, 1, 1, 0, 0, 0, 0,
datetime.timezone.utc))
self.assertTrue(DataFrame.equals(df, response))
def test_get_table_data_dataframe_succeeds(self):
......@@ -245,310 +488,603 @@ class TableUnitTest(unittest.TestCase):
exp = [{'id': 1, 'username': 'foo'}, {'id': 2, 'username': 'bar'}]
df = DataFrame.from_records(json.dumps(exp))
# mock
mock.get('/api/database/1/table/9/data', json=json.dumps(exp))
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
json=json.dumps(exp))
# test
response = RestClient().get_table_data(database_id=1, table_id=9)
response = RestClient().get_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
self.assertEqual(df.shape, response.shape)
self.assertTrue(DataFrame.equals(df, response))
def test_get_table_data_malformed_fails(self):
def test_get_table_data_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/table/9/data', status_code=400)
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=400)
# test
try:
response = RestClient().get_table_data(database_id=1, table_id=9)
response = RestClient().get_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except MalformedError:
pass
def test_get_table_data_not_allowed_fails(self):
def test_get_table_data_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/table/9/data', status_code=403)
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=403)
# test
try:
response = RestClient().get_table_data(database_id=1, table_id=9)
response = RestClient().get_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ForbiddenError:
pass
def test_get_table_data_not_found_fails(self):
def test_get_table_data_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/database/1/table/9/data', status_code=404)
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=404)
# test
try:
response = RestClient().get_table_data(database_id=1, table_id=9)
response = RestClient().get_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except NotExistsError:
pass
def test_get_table_data_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=503)
# test
try:
response = RestClient().get_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ServiceError:
pass
def test_get_table_data_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=202)
# test
try:
response = RestClient().get_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ResponseCodeError:
pass
def test_get_table_data_count_succeeds(self):
with requests_mock.Mocker() as mock:
exp = 2
# mock
mock.head('/api/database/1/table/9/data', headers={'X-Count': str(exp)})
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
headers={'X-Count': str(exp)})
# test
response = RestClient().get_table_data_count(database_id=1, table_id=9)
response = RestClient().get_table_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
timestamp=datetime.datetime(2024, 1, 1, 0, 0, 0, 0))
self.assertEqual(exp, response)
def test_get_table_data_count_malformed_fails(self):
def test_get_table_data_count_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head('/api/database/1/table/9/data', status_code=400)
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=400)
# test
try:
response = RestClient().get_table_data_count(database_id=1, table_id=9)
response = RestClient().get_table_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except MalformedError:
pass
def test_get_table_data_count_not_allowed_fails(self):
def test_get_table_data_count_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head('/api/database/1/table/9/data', status_code=403)
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=403)
# test
try:
response = RestClient().get_table_data_count(database_id=1, table_id=9)
response = RestClient().get_table_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ForbiddenError:
pass
def test_get_table_data_count_not_found_fails(self):
def test_get_table_data_count_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head('/api/database/1/table/9/data', status_code=404)
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=404)
# test
try:
response = RestClient().get_table_data_count(database_id=1, table_id=9)
response = RestClient().get_table_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except NotExistsError:
pass
def test_get_table_data_count_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=503)
# test
try:
response = RestClient().get_table_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ServiceError:
pass
def test_get_table_data_count_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=202)
# test
try:
response = RestClient().get_table_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ResponseCodeError:
pass
def test_get_table_data_count_not_countable_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.head('/api/database/1/table/9/data', status_code=409)
mock.head(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=409)
# test
try:
response = RestClient().get_table_data_count(database_id=1, table_id=9)
response = RestClient().get_table_data_count(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ExternalSystemError:
pass
def test_create_table_data_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table/9/data', status_code=201)
mock.post(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=201)
# test
client = RestClient(username="a", password="b")
client.create_table_data(database_id=1, table_id=9,
client.create_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'})
def test_create_table_data_malformed_fails(self):
def test_create_table_data_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table/9/data', status_code=400)
mock.post(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=400)
# test
try:
client = RestClient(username="a", password="b")
client.create_table_data(database_id=1, table_id=9,
client.create_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'})
except MalformedError:
pass
def test_create_table_data_not_allowed_fails(self):
def test_create_table_data_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table/9/data', status_code=403)
mock.post(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=403)
# test
try:
client = RestClient(username="a", password="b")
client.create_table_data(database_id=1, table_id=9,
client.create_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'})
except ForbiddenError:
pass
def test_create_table_data_not_found_fails(self):
def test_create_table_data_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post('/api/database/1/table/9/data', status_code=404)
mock.post(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=404)
# test
try:
client = RestClient(username="a", password="b")
client.create_table_data(database_id=1, table_id=9,
client.create_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'})
except NotExistsError:
pass
def test_create_table_data_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=503)
# test
try:
client = RestClient(username="a", password="b")
client.create_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'})
except ServiceError:
pass
def test_create_table_data_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=200)
# test
try:
client = RestClient(username="a", password="b")
client.create_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'})
except ResponseCodeError:
pass
def test_create_table_data_anonymous_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.post(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=503)
# test
try:
RestClient().create_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'})
except AuthenticationError:
pass
def test_update_table_data_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/1/table/9/data', status_code=202)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=202)
# test
client = RestClient(username="a", password="b")
client.update_table_data(database_id=1, table_id=9,
client.update_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'},
keys={'id': 1})
def test_update_table_data_malformed_fails(self):
def test_update_table_data_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/1/table/9/data', status_code=400)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=400)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_data(database_id=1, table_id=9,
client.update_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'},
keys={'id': 1})
except MalformedError:
pass
def test_update_table_data_not_allowed_fails(self):
def test_update_table_data_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/1/table/9/data', status_code=403)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=403)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_data(database_id=1, table_id=9,
client.update_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'},
keys={'id': 1})
except ForbiddenError:
pass
def test_update_table_data_not_found_fails(self):
def test_update_table_data_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/1/table/9/data', status_code=404)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=404)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_data(database_id=1, table_id=9,
client.update_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'},
keys={'id': 1})
except NotExistsError:
pass
def test_update_table_data_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=503)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'},
keys={'id': 1})
except ServiceError:
pass
def test_update_table_data_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=200)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
data={'name': 'Josiah', 'age': 45, 'gender': 'male'},
keys={'id': 1})
except ResponseCodeError:
pass
def test_delete_table_data_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/9/data', status_code=202)
mock.delete(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=202)
# test
client = RestClient(username="a", password="b")
client.delete_table_data(database_id=1, table_id=9, keys={'id': 1})
client.delete_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692", keys={'id': 1})
def test_delete_table_data_malformed_fails(self):
def test_delete_table_data_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/9/data', status_code=400)
mock.delete(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=400)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table_data(database_id=1, table_id=9, keys={'id': 1})
client.delete_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692", keys={'id': 1})
except MalformedError:
pass
def test_delete_table_data_not_allowed_fails(self):
def test_delete_table_data_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/9/data', status_code=403)
mock.delete(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=403)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table_data(database_id=1, table_id=9, keys={'id': 1})
client.delete_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692", keys={'id': 1})
except ForbiddenError:
pass
def test_delete_table_data_not_found_fails(self):
def test_delete_table_data_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/9/data', status_code=404)
mock.delete(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=404)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table_data(database_id=1, table_id=9, keys={'id': 1})
client.delete_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692", keys={'id': 1})
except NotExistsError:
pass
def test_delete_table_data_not_auth_fails(self):
def test_delete_table_data_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=503)
# test
try:
client = RestClient(username="a", password="b")
client.delete_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692", keys={'id': 1})
except ServiceError:
pass
def test_delete_table_data_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete('/api/database/1/table/9/data', status_code=404)
mock.delete(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=200)
# test
try:
RestClient().delete_table_data(database_id=1, table_id=9, keys={'id': 1})
client = RestClient(username="a", password="b")
client.delete_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692", keys={'id': 1})
except ResponseCodeError:
pass
def test_delete_table_data_anonymous_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.delete(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/data',
status_code=404)
# test
try:
RestClient().delete_table_data(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
keys={'id': 1})
except AuthenticationError:
pass
def test_update_table_column_succeeds(self):
with requests_mock.Mocker() as mock:
exp = Column(id=1,
exp = Column(id="2f30858d-b47b-4068-9028-fb0524ddf6cb",
ord=0,
name="ID",
database_id=1,
table_id=2,
database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
internal_name="id",
auto_generated=True,
is_primary_key=True,
type=ColumnType.BIGINT,
is_public=True,
concept=ConceptBrief(id=2,
concept=ConceptBrief(id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
uri="http://dbpedia.org/page/Category:Precipitation",
name="Precipitation"),
unit=UnitBrief(id=2,
unit=UnitBrief(id="029d773f-f98b-40c0-ab22-b8b1635d4fbc",
uri="http://www.wikidata.org/entity/Q119856947",
name="liters per square meter"),
is_null_allowed=False)
# mock
mock.put('/api/database/1/table/2/column/1', json=exp.model_dump(), status_code=202)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/column/758da021-f809-4d87-a866-9d4664a039dc',
json=exp.model_dump(),
status_code=202)
# test
client = RestClient(username="a", password="b")
response = client.update_table_column(database_id=1, table_id=2, column_id=1,
response = client.update_table_column(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
column_id="758da021-f809-4d87-a866-9d4664a039dc",
unit_uri="http://www.wikidata.org/entity/Q119856947",
concept_uri="http://dbpedia.org/page/Category:Precipitation")
self.assertEqual(exp, response)
def test_update_table_column_malformed_fails(self):
def test_update_table_column_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/1/table/2/column/1', status_code=400)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/column/758da021-f809-4d87-a866-9d4664a039dc',
status_code=400)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_column(database_id=1, table_id=2, column_id=1,
client.update_table_column(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
column_id="758da021-f809-4d87-a866-9d4664a039dc",
unit_uri="http://www.wikidata.org/entity/Q119856947",
concept_uri="http://dbpedia.org/page/Category:Precipitation")
except MalformedError:
pass
def test_update_table_column_not_allowed_fails(self):
def test_update_table_column_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/1/table/2/column/1', status_code=403)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/column/758da021-f809-4d87-a866-9d4664a039dc',
status_code=403)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_column(database_id=1, table_id=2, column_id=1,
client.update_table_column(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
column_id="758da021-f809-4d87-a866-9d4664a039dc",
unit_uri="http://www.wikidata.org/entity/Q119856947",
concept_uri="http://dbpedia.org/page/Category:Precipitation")
except ForbiddenError:
pass
def test_update_table_column_not_found_fails(self):
def test_update_table_column_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/1/table/2/column/1', status_code=404)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/column/758da021-f809-4d87-a866-9d4664a039dc',
status_code=404)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_column(database_id=1, table_id=2, column_id=1,
client.update_table_column(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
column_id="758da021-f809-4d87-a866-9d4664a039dc",
unit_uri="http://www.wikidata.org/entity/Q119856947",
concept_uri="http://dbpedia.org/page/Category:Precipitation")
except NotExistsError:
pass
def test_update_table_column_not_auth_fails(self):
def test_update_table_column_502_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/column/758da021-f809-4d87-a866-9d4664a039dc',
status_code=502)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_column(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
column_id="758da021-f809-4d87-a866-9d4664a039dc",
unit_uri="http://www.wikidata.org/entity/Q119856947",
concept_uri="http://dbpedia.org/page/Category:Precipitation")
except ServiceConnectionError:
pass
def test_update_table_column_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/column/758da021-f809-4d87-a866-9d4664a039dc',
status_code=503)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_column(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
column_id="758da021-f809-4d87-a866-9d4664a039dc",
unit_uri="http://www.wikidata.org/entity/Q119856947",
concept_uri="http://dbpedia.org/page/Category:Precipitation")
except ServiceError:
pass
def test_update_table_column_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/column/758da021-f809-4d87-a866-9d4664a039dc',
status_code=200)
# test
try:
client = RestClient(username="a", password="b")
client.update_table_column(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
column_id="758da021-f809-4d87-a866-9d4664a039dc",
unit_uri="http://www.wikidata.org/entity/Q119856947",
concept_uri="http://dbpedia.org/page/Category:Precipitation")
except ResponseCodeError:
pass
def test_update_table_column_anonymous_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('/api/database/1/table/2/column/1', status_code=404)
mock.put(
'/api/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/column/758da021-f809-4d87-a866-9d4664a039dc',
status_code=404)
# test
try:
RestClient().update_table_column(database_id=1, table_id=2, column_id=1,
RestClient().update_table_column(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692",
column_id="758da021-f809-4d87-a866-9d4664a039dc",
unit_uri="http://www.wikidata.org/entity/Q119856947",
concept_uri="http://dbpedia.org/page/Category:Precipitation")
except AuthenticationError:
......@@ -559,31 +1095,79 @@ class TableUnitTest(unittest.TestCase):
exp = TableStatistics(
columns={"id": ColumnStatistic(val_min=1.0, val_max=9.0, mean=5.0, median=5.0, std_dev=2.73)})
# mock
mock.get('/api/analyse/database/1/table/2/statistics', json=exp.model_dump(), status_code=202)
mock.get(
'/api/analyse/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/statistics',
json=exp.model_dump(), status_code=202)
# test
response = RestClient().analyse_table_statistics(database_id=1, table_id=2)
response = RestClient().analyse_table_statistics(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
self.assertEqual(exp, response)
def test_analyse_table_statistics_malformed_fails(self):
def test_analyse_table_statistics_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/analyse/database/1/table/2/statistics', status_code=400)
mock.get(
'/api/analyse/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/statistics',
status_code=400)
# test
try:
RestClient().analyse_table_statistics(database_id=1, table_id=2)
RestClient().analyse_table_statistics(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except MalformedError:
pass
def test_analyse_table_statistics_not_found_fails(self):
def test_analyse_table_statistics_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/analyse/database/1/table/2/statistics', status_code=404)
mock.get(
'/api/analyse/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/statistics',
status_code=404)
# test
try:
RestClient().analyse_table_statistics(database_id=1, table_id=2)
RestClient().analyse_table_statistics(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except NotExistsError:
pass
def test_analyse_table_statistics_502_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/analyse/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/statistics',
status_code=502)
# test
try:
RestClient().analyse_table_statistics(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ServiceConnectionError:
pass
def test_analyse_table_statistics_503_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/analyse/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/statistics',
status_code=503)
# test
try:
RestClient().analyse_table_statistics(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ServiceError:
pass
def test_analyse_table_statistics_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get(
'/api/analyse/database/6bd39359-b154-456d-b9c2-caa516a45732/table/b3230b86-4743-498d-9015-3fad58049692/statistics',
status_code=200)
# test
try:
RestClient().analyse_table_statistics(database_id="6bd39359-b154-456d-b9c2-caa516a45732",
table_id="b3230b86-4743-498d-9015-3fad58049692")
except ResponseCodeError:
pass
if __name__ == "__main__":
unittest.main()
lib/python/tests/test_unit_unit.py 0 → 100644
View file @ 841d4894
import unittest
import requests_mock
from dbrepo.RestClient import RestClient
from dbrepo.api.dto import UnitBrief
from dbrepo.api.exceptions import ResponseCodeError
class UserUnitTest(unittest.TestCase):
def test_get_units_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [UnitBrief(id="d8eec1ab-7b37-4eb3-bdf7-b44a8b384c0b",
uri='http://www.ontology-of-units-of-measure.org/resource/om-2/CelsiusTemperature',
name='Celsius Temperature')]
# mock
mock.get('/api/unit', json=[exp[0].model_dump()])
# test
response = RestClient().get_units()
self.assertEqual(exp, response)
def test_get_units_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('/api/unit', status_code=202)
# test
try:
RestClient().get_units()
except ResponseCodeError:
pass
if __name__ == "__main__":
unittest.main()
lib/python/tests/test_unit_user.py
View file @ 841d4894
......@@ -3,9 +3,9 @@ import unittest
import requests_mock
from dbrepo.RestClient import RestClient
from dbrepo.api.dto import User, UserAttributes, UserBrief
from dbrepo.api.exceptions import ResponseCodeError, UsernameExistsError, EmailExistsError, NotExistsError, \
ForbiddenError, AuthenticationError, MalformedError, ServiceError
from dbrepo.api.dto import UserAttributes, UserBrief, User, UnitBrief
from dbrepo.api.exceptions import ResponseCodeError, NotExistsError, \
ForbiddenError, AuthenticationError, ServiceError, MalformedError
class UserUnitTest(unittest.TestCase):
......@@ -22,101 +22,45 @@ class UserUnitTest(unittest.TestCase):
def test_get_users_empty_succeeds(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('http://localhost/api/user', json=[])
mock.get('/api/user', json=[])
# test
response = RestClient().get_users()
self.assertEqual([], response)
def test_get_user_succeeds(self):
def test_get_users_succeeds(self):
with requests_mock.Mocker() as mock:
exp = [
User(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise',
attributes=UserAttributes(theme='dark'))
]
exp = [UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise')]
# mock
mock.get('http://localhost/api/user', json=[exp[0].model_dump()])
mock.get('/api/user', json=[exp[0].model_dump()])
# test
response = RestClient().get_users()
self.assertEqual(exp, response)
def test_get_user_fails(self):
def test_get_users_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('http://localhost/api/user', status_code=404)
mock.get('/api/user', status_code=404)
# test
try:
response = RestClient().get_users()
except ResponseCodeError as e:
pass
def test_create_user_succeeds(self):
with requests_mock.Mocker() as mock:
exp = UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise')
# mock
mock.post('http://localhost/api/user', json=exp.model_dump(), status_code=201)
# test
response = RestClient().create_user(username='mweise', password='s3cr3t', email='mweise@example.com')
self.assertEqual(exp, response)
def test_create_user_bad_request_fails(self):
with requests_mock.Mocker() as mock:
exp = UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise')
# mock
mock.post('http://localhost/api/user', json=exp.model_dump(), status_code=400)
# test
try:
response = RestClient().create_user(username='mweise', password='s3cr3t', email='mweise@example.com')
except MalformedError as e:
pass
def test_create_user_username_exists_fails(self):
with requests_mock.Mocker() as mock:
exp = UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise')
# mock
mock.post('http://localhost/api/user', json=exp.model_dump(), status_code=409)
# test
try:
response = RestClient().create_user(username='mweise', password='s3cr3t', email='mweise@example.com')
except UsernameExistsError as e:
pass
def test_create_user_default_role_not_exists_fails(self):
with requests_mock.Mocker() as mock:
exp = UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise')
# mock
mock.post('http://localhost/api/user', json=exp.model_dump(), status_code=404)
# test
try:
response = RestClient().create_user(username='mweise', password='s3cr3t', email='mweise@example.com')
except NotExistsError as e:
pass
def test_create_user_emails_exists_fails(self):
with requests_mock.Mocker() as mock:
exp = UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise')
# mock
mock.post('http://localhost/api/user', json=exp.model_dump(), status_code=417)
# test
try:
response = RestClient().create_user(username='mweise', password='s3cr3t', email='mweise@example.com')
except EmailExistsError as e:
pass
def test_get_user_succeeds(self):
with requests_mock.Mocker() as mock:
exp = User(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise',
attributes=UserAttributes(theme='dark'))
attributes=UserAttributes(theme='dark', language='en'))
# mock
mock.get('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16',
mock.get('/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16',
json=exp.model_dump())
# test
response = RestClient().get_user(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16')
self.assertEqual(exp, response)
def test_get_user_not_found_fails(self):
def test_get_user_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.get('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16', status_code=404)
mock.get('/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16', status_code=404)
# test
try:
response = RestClient().get_user(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16')
......@@ -125,10 +69,9 @@ class UserUnitTest(unittest.TestCase):
def test_update_user_succeeds(self):
with requests_mock.Mocker() as mock:
exp = UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise', given_name='Martin',
attributes=UserAttributes(theme='dark'))
exp = UserBrief(id='8638c043-5145-4be8-a3e4-4b79991b0a16', username='mweise', given_name='Martin')
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16', status_code=202,
mock.put('/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16', status_code=202,
json=exp.model_dump())
# test
client = RestClient(username="a", password="b")
......@@ -136,95 +79,88 @@ class UserUnitTest(unittest.TestCase):
language='en', theme='light')
self.assertEqual(exp, response)
def test_update_user_not_allowed_fails(self):
def test_get_user_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16', status_code=403)
mock.get(f'/api/user/f27921d4-b05f-4e21-a122-4953a6a779a2', status_code=403)
# test
try:
client = RestClient(username="a", password="b")
response = client.update_user(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16', firstname='Martin',
language='en', theme='light')
except ForbiddenError as e:
RestClient().get_user('f27921d4-b05f-4e21-a122-4953a6a779a2')
except ForbiddenError:
pass
def test_update_user_not_found_fails(self):
def test_get_user_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16', status_code=404)
mock.get(f'/api/user/f27921d4-b05f-4e21-a122-4953a6a779a2', status_code=404)
# test
try:
client = RestClient(username="a", password="b")
response = client.update_user(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16', firstname='Martin',
language='en', theme='light')
except NotExistsError as e:
RestClient().get_user('f27921d4-b05f-4e21-a122-4953a6a779a2')
except NotExistsError:
pass
def test_update_user_not_auth_fails(self):
def test_get_user_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16', status_code=405)
mock.get(f'/api/user/f27921d4-b05f-4e21-a122-4953a6a779a2', status_code=409)
# test
try:
response = RestClient().update_user(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16', firstname='Martin',
language='en', theme='light')
except AuthenticationError as e:
RestClient().get_user('f27921d4-b05f-4e21-a122-4953a6a779a2')
except ResponseCodeError:
pass
def test_update_user_password_succeeds(self):
def test_update_user_anonymous_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16/password', status_code=202)
mock.put('/api/user/f27921d4-b05f-4e21-a122-4953a6a779a2', status_code=202)
# test
client = RestClient(username="a", password="b")
response = client.update_user_password(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16',
password='s3cr3t1n0rm4t10n')
try:
RestClient().update_user(user_id='f27921d4-b05f-4e21-a122-4953a6a779a2', theme='dark', language='en')
except AuthenticationError:
pass
def test_update_user_password_not_allowed_fails(self):
def test_update_user_400_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16/password', status_code=403)
mock.put('/api/user/f27921d4-b05f-4e21-a122-4953a6a779a2', status_code=400)
# test
try:
client = RestClient(username="a", password="b")
response = client.update_user_password(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16',
password='s3cr3t1n0rm4t10n')
except ForbiddenError as e:
RestClient(username='foo', password='bar').update_user(user_id='f27921d4-b05f-4e21-a122-4953a6a779a2',
theme='dark', language='en')
except MalformedError:
pass
def test_update_user_password_not_found_fails(self):
def test_update_user_403_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16/password', status_code=404)
mock.put('/api/user/f27921d4-b05f-4e21-a122-4953a6a779a2', status_code=403)
# test
try:
client = RestClient(username="a", password="b")
response = client.update_user_password(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16',
password='s3cr3t1n0rm4t10n')
except NotExistsError as e:
RestClient(username='foo', password='bar').update_user(user_id='f27921d4-b05f-4e21-a122-4953a6a779a2',
theme='dark', language='en')
except ForbiddenError:
pass
def test_update_user_password_keycloak_fails(self):
def test_update_user_404_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16/password', status_code=503)
mock.put('/api/user/f27921d4-b05f-4e21-a122-4953a6a779a2', status_code=404)
# test
try:
client = RestClient(username="a", password="b")
response = client.update_user_password(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16',
password='s3cr3t1n0rm4t10n')
except ServiceError as e:
RestClient(username='foo', password='bar').update_user(user_id='f27921d4-b05f-4e21-a122-4953a6a779a2',
theme='dark', language='en')
except NotExistsError:
pass
def test_update_user_password_not_auth_fails(self):
def test_update_user_unknown_fails(self):
with requests_mock.Mocker() as mock:
# mock
mock.put('http://localhost/api/user/8638c043-5145-4be8-a3e4-4b79991b0a16/password', status_code=503)
mock.put('/api/user/f27921d4-b05f-4e21-a122-4953a6a779a2', status_code=200)
# test
try:
response = RestClient().update_user_password(user_id='8638c043-5145-4be8-a3e4-4b79991b0a16',
password='s3cr3t1n0rm4t10n')
except AuthenticationError as e:
RestClient(username='foo', password='bar').update_user(user_id='f27921d4-b05f-4e21-a122-4953a6a779a2',
theme='dark', language='en')
except ResponseCodeError:
pass
......