diff --git a/CHANGE.txt b/CHANGE.txt index b6a6d4a..0349142 100644 --- a/CHANGE.txt +++ b/CHANGE.txt @@ -2,6 +2,13 @@ LabKey Python Client API News +++++++++++ +What's New in the LabKey 4.2.0 package +============================== + +*Release date: 02/26/2026* +- Add get_queries to query module + - Accessible via API wrapper as api.query.get_queries + What's New in the LabKey 4.1.0 package ============================== diff --git a/labkey/query.py b/labkey/query.py index 24a7160..385c06b 100644 --- a/labkey/query.py +++ b/labkey/query.py @@ -40,11 +40,12 @@ ############################################################################ """ + import functools from typing import List, Literal, NotRequired, TextIO, TypedDict from .server_context import ServerContext -from .utils import waf_encode +from .utils import waf_encode, transform_options _default_timeout = 60 * 5 # 5 minutes @@ -60,6 +61,7 @@ class Pagination: ALL = "all" NONE = "none" + # TODO: Provide filter generators. # # There are some inconsistencies between the different filter types with multiple values, @@ -143,7 +145,6 @@ class Types: ARRAY_ISEMPTY = "arrayisempty" ARRAY_ISNOTEMPTY = "arrayisnotempty" - # Table/Query-wise operators Q = "q" @@ -704,6 +705,53 @@ def move_rows( ) +get_queries_fields = [ + "schema_name", + "include_columns", + "include_system_queries", + "include_title", + "include_user_queries", + "include_view_data_url", + "query_detail_columns", +] + + +def get_queries( + server_context: ServerContext, + schema_name: str, + container_path: str = None, + timeout=_default_timeout, + **kwargs, +) -> dict: + """ + :param server_context: A LabKey server context. See utils.create_server_context. + :param schema_name: schema of table + :param container_path: folder path if not already part of server_context + :param timeout: Request timeout in seconds (defaults to 300s) + :param kwargs: Optional parameters supported by this API: + include_columns: boolean, if set to False, information about the available columns in this query will not be + included in the results. Default is True. + include_system_queries: boolean, if set to false, system-defined queries will not be included in the results. + Default is True. + include_title: boolean, if set to False, no custom query titles will be included. Instead, titles will be + identical to names. Default is True. + include_user_queries: boolean, if set to False, user-defined queries will not be included in the results. + Default is True. + include_view_data_url: boolean, if set to False, view data URLs will not be included in the results. + Default is True. + query_detail_columns: boolean, if set to True, and includeColumns is set to True, information about the + available columns will be the same details as specified by getQueryDetails for columns. Defaults to False. + :return: dict + """ + url = server_context.build_url("query", "getQueries.api", container_path=container_path) + payload = {"schemaName": schema_name} + + if len(kwargs) > 0: + payload = {**payload, **transform_options(kwargs, get_queries_fields)} + + return server_context.make_request(url, payload, timeout=timeout) + + class QueryWrapper: """ Wrapper for all of the API methods exposed in the query module. Used by the APIWrapper class. @@ -939,3 +987,29 @@ def move_rows( audit_user_comment, timeout, ) + + @functools.wraps(get_queries) + def get_queries( + self, + schema_name: str, + container_path: str = None, + include_columns: bool = None, + include_system_queries: bool = None, + include_title: bool = None, + include_user_queries: bool = None, + include_view_data_url: bool = None, + query_detail_columns: bool = None, + timeout=_default_timeout, + ): + return get_queries( + self.server_context, + schema_name, + container_path, + timeout, + include_columns=include_columns, + include_system_queries=include_system_queries, + include_title=include_title, + include_user_queries=include_user_queries, + include_view_data_url=include_view_data_url, + query_detail_columns=query_detail_columns, + ) diff --git a/labkey/server_context.py b/labkey/server_context.py index 7de7322..a5eca44 100644 --- a/labkey/server_context.py +++ b/labkey/server_context.py @@ -79,8 +79,6 @@ def __init__( self._session = requests.Session() self._session.headers.update({"User-Agent": f"LabKey Python API/{client_version}"}) - print(f"User Agent header: LabKey Python API/{client_version}") - if self._use_ssl: self._scheme = "https://" if not self._verify_ssl: diff --git a/labkey/storage.py b/labkey/storage.py index bc6c22c..686c8dd 100644 --- a/labkey/storage.py +++ b/labkey/storage.py @@ -46,6 +46,7 @@ ############################################################################ """ + import functools from dataclasses import dataclass diff --git a/labkey/utils.py b/labkey/utils.py index c87859c..4cded31 100644 --- a/labkey/utils.py +++ b/labkey/utils.py @@ -17,6 +17,7 @@ from functools import wraps from datetime import date, datetime from base64 import b64encode +from typing import List from urllib import parse @@ -91,3 +92,34 @@ def waf_encode(value: str) -> str: if value: return "/*{{base64/x-www-form-urlencoded/wafText}}*/" + btoa(encode_uri_component(value)) return value + + +def snake_to_camel(value: str): + """ + Converts a snake_case string to camelCase + """ + if not value: + return value + + if "_" not in value: + return value + + parts = [part for part in value.split("_") if part] + + if len(parts) == 0: + return "" + + return parts[0].lower() + "".join([part.title() for part in parts[1:]]) + + +def transform_options(options: dict, expected_keys: List[str]) -> dict: + """ + Converts a dict with snake_case keys to a new dict with camelCase keys, only copying keys from expected_keys + """ + transformed_options = {} + + for key, item in options.items(): + if key in expected_keys: + transformed_options[snake_to_camel(key)] = item + + return transformed_options diff --git a/pyproject.toml b/pyproject.toml index 8fb88bb..66baf60 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name="labkey" -version = "4.1.0" +version = "4.2.0" description = "Python client API for LabKey Server" dependencies = ["requests>=2.32.5"] readme = "README.md" diff --git a/samples/query_examples.py b/samples/query_examples.py index 071048b..b081239 100644 --- a/samples/query_examples.py +++ b/samples/query_examples.py @@ -16,6 +16,7 @@ """ Examples using the Query.py API """ + from labkey.api_wrapper import APIWrapper from labkey.exceptions import ( RequestError, diff --git a/test/integration/test_query.py b/test/integration/test_query.py index e6a3db3..51953da 100644 --- a/test/integration/test_query.py +++ b/test/integration/test_query.py @@ -346,8 +346,8 @@ def test_api_save_rows(api: APIWrapper, blood_sample_type_fixture, tissue_sample assert resp["committed"] == False assert resp["errorCount"] == 1 assert ( - "SampleID or Name is required for sample on row 2" in - resp["result"][0]["errors"]["exception"] + "SampleID or Name is required for sample on row 2" + in resp["result"][0]["errors"]["exception"] ) # Fix the first command by specifying a name for the sample @@ -411,3 +411,45 @@ def test_api_save_rows(api: APIWrapper, blood_sample_type_fixture, tissue_sample assert resp["result"][2]["rowsAffected"] == 1 assert resp["result"][2]["rows"][0]["rowid"] == first_tissue_row_id assert resp["result"][2]["rows"][0]["receiveddate"] == "2025-07-07 12:34:56.000" + + +expected_fields = { + "canEdit", + "canEditSharedViews", + "columns", + "hidden", + "inherit", + "isIncludedForLookups", + "isInherited", + "isMetadataOverrideable", + "isUserDefined", + "moduleName", + "name", + "snapshot", + "title", + "viewDataUrl", +} + + +def test_get_queries(api: APIWrapper): + resp = api.query.get_queries("core") + + all_queries_count = len(resp["queries"]) + assert set(resp.keys()) == {"schemaName", "queries"} + assert resp["schemaName"] == "core" + assert all_queries_count > 0 + assert set(resp["queries"][0].keys()) == set(expected_fields) + + resp = api.query.get_queries("core", include_system_queries=False, include_user_queries=False) + + assert set(resp.keys()) == {"schemaName", "queries"} + assert resp["schemaName"] == "core" + # By excluding system queries, and user queries, we should have no queries + assert len(resp["queries"]) == 0 + + resp = api.query.get_queries("core", include_columns=False, include_view_data_url=False) + + assert set(resp["queries"][0].keys()) == expected_fields - { + "columns", + "viewDataUrl", + } diff --git a/test/unit/test_domain.py b/test/unit/test_domain.py index 7ddbe67..ea566db 100644 --- a/test/unit/test_domain.py +++ b/test/unit/test_domain.py @@ -16,9 +16,7 @@ import json import os import tempfile -import unittest - -import unittest.mock as mock +import pytest from labkey.domain import ( create, @@ -42,214 +40,176 @@ throws_error_test_get, ) - domain_controller = "property" -class TestCreate(unittest.TestCase): - def setUp(self): - - domain_definition = { - "kind": "IntList", - "domainDesign": { - "name": "TheTestList", - "fields": [{"name": "theKey", "rangeURI": "int"}], - }, - "options": {"keyName": "theKey"}, - } - - class MockCreate(MockLabKey): - api = "createDomain.api" - default_action = domain_controller - default_success_body = domain_definition - - self.service = MockCreate() +@pytest.fixture +def create_setup(): + domain_definition = { + "kind": "IntList", + "domainDesign": { + "name": "TheTestList", + "fields": [{"name": "theKey", "rangeURI": "int"}], + }, + "options": {"keyName": "theKey"}, + } + + class MockCreate(MockLabKey): + api = "createDomain.api" + default_action = domain_controller + default_success_body = domain_definition + + service = MockCreate() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": json.dumps(domain_definition, sort_keys=True), + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), domain_definition] + return service, args, expected_kwargs + + +def test_create_success(create_setup): + service, args, expected_kwargs = create_setup + success_test(service.get_successful_response(), create, False, *args, **expected_kwargs) + + +def test_create_unauthorized(create_setup): + service, args, expected_kwargs = create_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + create, + *args, + **expected_kwargs, + ) - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": json.dumps(domain_definition, sort_keys=True), - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } - self.args = [mock_server_context(self.service), domain_definition] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - create, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - create, - *self.args, - **self.expected_kwargs - ) - - -class TestDrop(unittest.TestCase): +@pytest.fixture +def drop_setup(): schema_name = "lists" query_name = "TheTestList" - def setUp(self): - class MockDrop(MockLabKey): - api = "deleteDomain.api" - default_action = domain_controller - default_success_body = {} - - self.service = MockDrop() - - payload = {"schemaName": self.schema_name, "queryName": self.query_name} + class MockDrop(MockLabKey): + api = "deleteDomain.api" + default_action = domain_controller + default_success_body = {} + + service = MockDrop() + payload = {"schemaName": schema_name, "queryName": query_name} + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": json.dumps(payload, sort_keys=True), + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema_name, query_name] + return service, args, expected_kwargs + + +def test_drop_success(drop_setup): + service, args, expected_kwargs = drop_setup + success_test(service.get_successful_response(), drop, False, *args, **expected_kwargs) + + +def test_drop_unauthorized(drop_setup): + service, args, expected_kwargs = drop_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + drop, + *args, + **expected_kwargs, + ) - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": json.dumps(payload, sort_keys=True), - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } - self.args = [ - mock_server_context(self.service), - self.schema_name, - self.query_name, - ] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - drop, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - drop, - *self.args, - **self.expected_kwargs - ) - - -class TestGet(unittest.TestCase): +@pytest.fixture +def get_setup(): schema_name = "lists" query_name = "TheTestList" - def setUp(self): - class MockGet(MockLabKey): - api = "getDomain.api" - default_action = domain_controller - default_success_body = {} + class MockGet(MockLabKey): + api = "getDomain.api" + default_action = domain_controller + default_success_body = {} + + service = MockGet() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "headers": None, + "params": {"schemaName": schema_name, "queryName": query_name}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema_name, query_name] + return service, args, expected_kwargs + + +def test_get_success(get_setup): + service, args, expected_kwargs = get_setup + success_test_get(service.get_successful_response(), get, False, *args, **expected_kwargs) + + +def test_get_unauthorized(get_setup): + service, args, expected_kwargs = get_setup + throws_error_test_get( + RequestAuthorizationError, + service.get_unauthorized_response(), + get, + *args, + **expected_kwargs, + ) - self.service = MockGet() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "headers": None, - "params": {"schemaName": self.schema_name, "queryName": self.query_name}, - "timeout": 300, - "allow_redirects": False, - } +@pytest.fixture +def infer_fields_setup(): + class MockInferFields(MockLabKey): + api = "inferDomain.api" + default_action = domain_controller + default_success_body = {} + + service = MockInferFields() + fd, path = tempfile.mkstemp() + with os.fdopen(fd, "w") as tmp: + tmp.write("Name\tAge\nNick\t32\nBrian\t27\n") + + # Re-open the file for reading in the test + file = open(path, "r") + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": None, + "files": {"inferfile": file}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), file] + yield service, args, expected_kwargs + file.close() + os.remove(path) + + +def test_infer_fields_success(infer_fields_setup): + service, args, expected_kwargs = infer_fields_setup + success_test(service.get_successful_response(), infer_fields, False, *args, **expected_kwargs) + + +def test_infer_fields_unauthorized(infer_fields_setup): + service, args, expected_kwargs = infer_fields_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + infer_fields, + *args, + **expected_kwargs, + ) - self.args = [ - mock_server_context(self.service), - self.schema_name, - self.query_name, - ] - - def test_success(self): - test = self - success_test_get( - test, - self.service.get_successful_response(), - get, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test_get( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - get, - *self.args, - **self.expected_kwargs - ) - - -class TestInferFields(unittest.TestCase): - def setUp(self): - class MockInferFields(MockLabKey): - api = "inferDomain.api" - default_action = domain_controller - default_success_body = {} - - self.service = MockInferFields() - - self.fd, self.path = tempfile.mkstemp() - with os.fdopen(self.fd, "w") as tmp: - tmp.write("Name\tAge\nNick\t32\nBrian\t27\n") - self.file = tmp - - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": None, - "files": {"inferfile": self.file}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - self.args = [mock_server_context(self.service), self.file] - - def tearDown(self): - os.remove(self.path) - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - infer_fields, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - infer_fields, - *self.args, - **self.expected_kwargs - ) - - -class TestSave(unittest.TestCase): +@pytest.fixture +def save_setup(): domain = Domain( **{ "container": "TestContainer", @@ -260,230 +220,171 @@ class TestSave(unittest.TestCase): schema_name = "lists" query_name = "TheTestList" - def setUp(self): - class MockSave(MockLabKey): - api = "saveDomain.api" - default_action = domain_controller - default_success_body = {} - - self.service = MockSave() - - payload = { - "domainDesign": self.domain.to_json(), - "queryName": self.query_name, - "schemaName": self.schema_name, - } - - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": json.dumps(payload, sort_keys=True), - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [ - mock_server_context(self.service), - self.schema_name, - self.query_name, - self.domain, - ] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - save, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - save, - *self.args, - **self.expected_kwargs - ) - - -class TestConditionalFormatCreate(unittest.TestCase): - def setUp(self): - - self.domain_definition = { - "kind": "IntList", - "domainDesign": { - "name": "TheTestList_cf", - "fields": [ - { - "name": "theKey", - "rangeURI": "int", - "conditionalFormats": [ - { - "filter": encode_conditional_format_filter( - QueryFilter("theKey", 500) - ), - "textColor": "f44e3b", - "backgroundColor": "fcba03", - "bold": True, - "italic": True, - "strikethrough": False, - } - ], - } - ], - }, - } - - class MockCreate(MockLabKey): - api = "createDomain.api" - default_action = domain_controller - default_success_body = self.domain_definition + class MockSave(MockLabKey): + api = "saveDomain.api" + default_action = domain_controller + default_success_body = {} + + service = MockSave() + payload = { + "domainDesign": domain.to_json(), + "queryName": query_name, + "schemaName": schema_name, + } + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": json.dumps(payload, sort_keys=True), + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema_name, query_name, domain] + return service, args, expected_kwargs + + +def test_save_success(save_setup): + service, args, expected_kwargs = save_setup + success_test(service.get_successful_response(), save, False, *args, **expected_kwargs) + + +def test_save_unauthorized(save_setup): + service, args, expected_kwargs = save_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + save, + *args, + **expected_kwargs, + ) - self.service = MockCreate() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": json.dumps(self.domain_definition, sort_keys=True), - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } +@pytest.fixture +def conditional_format_create_setup(): + domain_definition = { + "kind": "IntList", + "domainDesign": { + "name": "TheTestList_cf", + "fields": [ + { + "name": "theKey", + "rangeURI": "int", + "conditionalFormats": [ + { + "filter": encode_conditional_format_filter(QueryFilter("theKey", 500)), + "textColor": "f44e3b", + "backgroundColor": "fcba03", + "bold": True, + "italic": True, + "strikethrough": False, + } + ], + } + ], + }, + } + + class MockCreate(MockLabKey): + api = "createDomain.api" + default_action = domain_controller + default_success_body = domain_definition + + service = MockCreate() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": json.dumps(domain_definition, sort_keys=True), + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), domain_definition] + return service, args, expected_kwargs + + +def test_conditional_format_create_success(conditional_format_create_setup): + service, args, expected_kwargs = conditional_format_create_setup + success_test(service.get_successful_response(), create, False, *args, **expected_kwargs) + + +def test_conditional_format_create_unauthorized(conditional_format_create_setup): + service, args, expected_kwargs = conditional_format_create_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + create, + *args, + **expected_kwargs, + ) - self.args = [mock_server_context(self.service), self.domain_definition] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - create, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - create, - *self.args, - **self.expected_kwargs - ) - - -class TestConditionalFormatSave(unittest.TestCase): +@pytest.fixture +def conditional_format_save_setup(): schema_name = "lists" query_name = "TheTestList_cf" - - def setUp(self): - self.test_domain = Domain( - **{ - "container": "TestContainer", - "description": "A Test Domain", - "domain_id": 5314, - "fields": [{"name": "theKey", "rangeURI": "int"}], - } - ) - - self.test_domain.fields[0].conditional_formats = [ - # create conditional format using our utility for a QueryFilter - conditional_format( - background_color="fcba03", - bold=True, - italic=True, - query_filter=QueryFilter("theKey", 200), - strike_through=True, - text_color="f44e3b", - ), - # create conditional format using our utility for a QueryFilter list - conditional_format( - background_color="fcba03", - bold=True, - italic=True, - query_filter=[ - QueryFilter("theKey", 500, QueryFilter.Types.GREATER_THAN), - QueryFilter("theKey", 1000, QueryFilter.Types.LESS_THAN), - ], - strike_through=True, - text_color="f44e3b", - ), - ] - - class MockSave(MockLabKey): - api = "saveDomain.api" - default_action = domain_controller - default_success_body = {} - - self.service = MockSave() - - payload = { - "domainDesign": self.test_domain.to_json(), - "queryName": self.query_name, - "schemaName": self.schema_name, - } - - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": json.dumps(payload, sort_keys=True), - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, + test_domain = Domain( + **{ + "container": "TestContainer", + "description": "A Test Domain", + "domain_id": 5314, + "fields": [{"name": "theKey", "rangeURI": "int"}], } - - self.args = [ - mock_server_context(self.service), - self.schema_name, - self.query_name, - self.test_domain, - ] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - save, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - save, - *self.args, - **self.expected_kwargs - ) - - -def suite(): - load_tests = unittest.TestLoader().loadTestsFromTestCase - return unittest.TestSuite( - [ - load_tests(TestCreate), - load_tests(TestDrop), - load_tests(TestGet), - load_tests(TestInferFields), - load_tests(TestSave), - load_tests(TestConditionalFormatCreate), - load_tests(TestConditionalFormatSave), - ] ) - - -if __name__ == "__main__": - unittest.main() + test_domain.fields[0].conditional_formats = [ + # create conditional format using our utility for a QueryFilter + conditional_format( + background_color="fcba03", + bold=True, + italic=True, + query_filter=QueryFilter("theKey", 200), + strike_through=True, + text_color="f44e3b", + ), + # create conditional format using our utility for a QueryFilter list + conditional_format( + background_color="fcba03", + bold=True, + italic=True, + query_filter=[ + QueryFilter("theKey", 500, QueryFilter.Types.GREATER_THAN), + QueryFilter("theKey", 1000, QueryFilter.Types.LESS_THAN), + ], + strike_through=True, + text_color="f44e3b", + ), + ] + + class MockSave(MockLabKey): + api = "saveDomain.api" + default_action = domain_controller + default_success_body = {} + + service = MockSave() + payload = { + "domainDesign": test_domain.to_json(), + "queryName": query_name, + "schemaName": schema_name, + } + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": json.dumps(payload, sort_keys=True), + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema_name, query_name, test_domain] + return service, args, expected_kwargs + + +def test_conditional_format_save_success(conditional_format_save_setup): + service, args, expected_kwargs = conditional_format_save_setup + success_test(service.get_successful_response(), save, True, *args, **expected_kwargs) + + +def test_conditional_format_save_unauthorized(conditional_format_save_setup): + service, args, expected_kwargs = conditional_format_save_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + save, + *args, + **expected_kwargs, + ) diff --git a/test/unit/test_experiment_api.py b/test/unit/test_experiment_api.py index 393032b..0c3b440 100644 --- a/test/unit/test_experiment_api.py +++ b/test/unit/test_experiment_api.py @@ -13,9 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. # -import unittest - -import unittest.mock as mock +import pytest from labkey.experiment import load_batch, save_batch, Batch, Run from labkey.exceptions import ( @@ -200,162 +198,132 @@ class MockSaveBatch(MockLabKey): batch_id = 54321 -class TestLoadBatch(unittest.TestCase): - def setUp(self): - self.service = MockLoadBatch() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": '{"assayId": 12345, "batchId": 54321}', - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), assay_id, batch_id] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - load_batch, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - load_batch, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - load_batch, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - load_batch, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - load_batch, - *self.args, - **self.expected_kwargs - ) - - -class TestSaveBatch(unittest.TestCase): - def setUp(self): - - data_rows = [] - - # Generate the Run object(s) - run = Run() - run.name = "python upload" - run.data_rows = data_rows - run.properties["RunFieldName"] = "Run Field Value" - - # Generate the Batch object(s) - batch = Batch() - batch.runs = [run] - batch.properties["PropertyName"] = "Property Value" - - self.service = MockSaveBatch() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": '{"assayId": 12345, "batches": [{"batchProtocolId": null, "comment": null, "created": null, "createdBy": null, "modified": null, "modifiedBy": null, "name": null, "properties": {"PropertyName": "Property Value"}, "runs": [{"name": "python upload", "properties": {"RunFieldName": "Run Field Value"}}]}]}', - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), assay_id, batch] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - save_batch, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - save_batch, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - save_batch, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - save_batch, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - save_batch, - *self.args, - **self.expected_kwargs - ) - - -def suite(): - load_tests = unittest.TestLoader().loadTestsFromTestCase - return unittest.TestSuite([load_tests(TestLoadBatch), load_tests(TestSaveBatch)]) - - -if __name__ == "__main__": - unittest.main() +@pytest.fixture +def load_batch_setup(): + service = MockLoadBatch() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": '{"assayId": 12345, "batchId": 54321}', + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), assay_id, batch_id] + return service, args, expected_kwargs + + +def test_load_batch_success(load_batch_setup): + service, args, expected_kwargs = load_batch_setup + success_test(service.get_successful_response(), load_batch, False, *args, **expected_kwargs) + + +def test_load_batch_unauthorized(load_batch_setup): + service, args, expected_kwargs = load_batch_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + load_batch, + *args, + **expected_kwargs, + ) + + +def test_load_batch_query_not_found(load_batch_setup): + service, args, expected_kwargs = load_batch_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + load_batch, + *args, + **expected_kwargs, + ) + + +def test_load_batch_server_not_found(load_batch_setup): + service, args, expected_kwargs = load_batch_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + load_batch, + *args, + **expected_kwargs, + ) + + +def test_load_batch_general_error(load_batch_setup): + service, args, expected_kwargs = load_batch_setup + throws_error_test( + RequestError, service.get_general_error_response(), load_batch, *args, **expected_kwargs + ) + + +@pytest.fixture +def save_batch_setup(): + data_rows = [] + + # Generate the Run object(s) + run = Run() + run.name = "python upload" + run.data_rows = data_rows + run.properties["RunFieldName"] = "Run Field Value" + + # Generate the Batch object(s) + batch = Batch() + batch.runs = [run] + batch.properties["PropertyName"] = "Property Value" + + service = MockSaveBatch() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": '{"assayId": 12345, "batches": [{"batchProtocolId": null, "comment": null, "created": null, "createdBy": null, "modified": null, "modifiedBy": null, "name": null, "properties": {"PropertyName": "Property Value"}, "runs": [{"name": "python upload", "properties": {"RunFieldName": "Run Field Value"}}]}]}', + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), assay_id, batch] + return service, args, expected_kwargs + + +def test_save_batch_success(save_batch_setup): + service, args, expected_kwargs = save_batch_setup + success_test(service.get_successful_response(), save_batch, False, *args, **expected_kwargs) + + +def test_save_batch_unauthorized(save_batch_setup): + service, args, expected_kwargs = save_batch_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + save_batch, + *args, + **expected_kwargs, + ) + + +def test_save_batch_query_not_found(save_batch_setup): + service, args, expected_kwargs = save_batch_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + save_batch, + *args, + **expected_kwargs, + ) + + +def test_save_batch_server_not_found(save_batch_setup): + service, args, expected_kwargs = save_batch_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + save_batch, + *args, + **expected_kwargs, + ) + + +def test_save_batch_general_error(save_batch_setup): + service, args, expected_kwargs = save_batch_setup + throws_error_test( + RequestError, service.get_general_error_response(), save_batch, *args, **expected_kwargs + ) diff --git a/test/unit/test_query_api.py b/test/unit/test_query_api.py index b474ad1..95ea87f 100644 --- a/test/unit/test_query_api.py +++ b/test/unit/test_query_api.py @@ -14,7 +14,7 @@ # limitations under the License. # import json -import unittest +import pytest from labkey.query import ( delete_rows, @@ -23,6 +23,7 @@ save_rows, select_rows, execute_sql, + get_queries, QueryFilter, ) from labkey.exceptions import ( @@ -61,610 +62,563 @@ class MockSaveRows(MockLabKey): api = "saveRows.api" +class MockGetQueries(MockLabKey): + api = "getQueries.api" + + schema = "testSchema" query = "testQuery" -class TestDeleteRows(unittest.TestCase): - def setUp(self): - self.service = MockDeleteRows() - - rows = "{id:1234}" - - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": '{"queryName": "' - + query - + '", "rows": "' - + rows - + '", "schemaName": "' - + schema - + '"}', - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } +@pytest.fixture +def delete_rows_setup(): + service = MockDeleteRows() + rows = "{id:1234}" + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": '{"queryName": "' + query + '", "rows": "' + rows + '", "schemaName": "' + schema + '"}', + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema, query, rows] + return service, args, expected_kwargs + + +def test_delete_rows_success(delete_rows_setup): + service, args, expected_kwargs = delete_rows_setup + success_test(service.get_successful_response(), delete_rows, True, *args, **expected_kwargs) + + +def test_delete_rows_unauthorized(delete_rows_setup): + service, args, expected_kwargs = delete_rows_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + delete_rows, + *args, + **expected_kwargs, + ) - self.args = [mock_server_context(self.service), schema, query, rows] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - delete_rows, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - delete_rows, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - delete_rows, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - delete_rows, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - delete_rows, - *self.args, - **self.expected_kwargs - ) - - -class TestUpdateRows(unittest.TestCase): - def setUp(self): - self.service = MockUpdateRows() - - rows = "{id:1234}" - - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": '{"queryName": "' - + query - + '", "rows": "' - + rows - + '", "schemaName": "' - + schema - + '"}', - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } - self.args = [mock_server_context(self.service), schema, query, rows] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - update_rows, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - update_rows, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - update_rows, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - update_rows, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - update_rows, - *self.args, - **self.expected_kwargs - ) - - -class TestInsertRows(unittest.TestCase): - def setUp(self): - self.service = MockInsertRows() - - rows = "{id:1234}" - - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": '{"queryName": "' - + query - + '", "rows": "' - + rows - + '", "schemaName": "' - + schema - + '"}', - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, - } +def test_delete_rows_query_not_found(delete_rows_setup): + service, args, expected_kwargs = delete_rows_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + delete_rows, + *args, + **expected_kwargs, + ) - self.args = [mock_server_context(self.service), schema, query, rows] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - insert_rows, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - insert_rows, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - insert_rows, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - insert_rows, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - insert_rows, - *self.args, - **self.expected_kwargs - ) - - -class TestExecuteSQL(unittest.TestCase): - def setUp(self): - self.service = MockExecuteSQL() - sql = "select * from " + schema + "." + query - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"sql": waf_encode(sql), "schemaName": schema}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - self.args = [mock_server_context(self.service), schema, sql] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - execute_sql, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - execute_sql, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - execute_sql, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - execute_sql, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - execute_sql, - *self.args, - **self.expected_kwargs - ) - - -class TestSelectRows(unittest.TestCase): - def setUp(self): - self.service = MockSelectRows() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"schemaName": schema, "query.queryName": query, "query.maxRows": -1}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } +def test_delete_rows_server_not_found(delete_rows_setup): + service, args, expected_kwargs = delete_rows_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + delete_rows, + *args, + **expected_kwargs, + ) - self.args = [mock_server_context(self.service), schema, query] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - select_rows, - True, - *self.args, - **self.expected_kwargs - ) - - def test_query_filter(self): - test = self - view_name = None - filter_array = [ - QueryFilter("Field1", "value", "eq"), - QueryFilter("Field2", "value1", "contains"), - QueryFilter("Field2", "value2", "contains"), - ] - args = list(self.args) + [view_name, filter_array] - # Expected query field values in post request body - query_field = { - "query.Field1~eq": ["value"], - "query.Field2~contains": ["value1", "value2"], + +def test_delete_rows_general_error(delete_rows_setup): + service, args, expected_kwargs = delete_rows_setup + throws_error_test( + RequestError, service.get_general_error_response(), delete_rows, *args, **expected_kwargs + ) + + +@pytest.fixture +def update_rows_setup(): + service = MockUpdateRows() + rows = "{id:1234}" + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": '{"queryName": "' + query + '", "rows": "' + rows + '", "schemaName": "' + schema + '"}', + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema, query, rows] + return service, args, expected_kwargs + + +def test_update_rows_success(update_rows_setup): + service, args, expected_kwargs = update_rows_setup + success_test(service.get_successful_response(), update_rows, True, *args, **expected_kwargs) + + +def test_update_rows_unauthorized(update_rows_setup): + service, args, expected_kwargs = update_rows_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + update_rows, + *args, + **expected_kwargs, + ) + + +def test_update_rows_query_not_found(update_rows_setup): + service, args, expected_kwargs = update_rows_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + update_rows, + *args, + **expected_kwargs, + ) + + +def test_update_rows_server_not_found(update_rows_setup): + service, args, expected_kwargs = update_rows_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + update_rows, + *args, + **expected_kwargs, + ) + + +def test_update_rows_general_error(update_rows_setup): + service, args, expected_kwargs = update_rows_setup + throws_error_test( + RequestError, service.get_general_error_response(), update_rows, *args, **expected_kwargs + ) + + +@pytest.fixture +def insert_rows_setup(): + service = MockInsertRows() + rows = "{id:1234}" + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": '{"queryName": "' + query + '", "rows": "' + rows + '", "schemaName": "' + schema + '"}', + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema, query, rows] + return service, args, expected_kwargs + + +def test_insert_rows_success(insert_rows_setup): + service, args, expected_kwargs = insert_rows_setup + success_test(service.get_successful_response(), insert_rows, True, *args, **expected_kwargs) + + +def test_insert_rows_unauthorized(insert_rows_setup): + service, args, expected_kwargs = insert_rows_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + insert_rows, + *args, + **expected_kwargs, + ) + + +def test_insert_rows_query_not_found(insert_rows_setup): + service, args, expected_kwargs = insert_rows_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + insert_rows, + *args, + **expected_kwargs, + ) + + +def test_insert_rows_server_not_found(insert_rows_setup): + service, args, expected_kwargs = insert_rows_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + insert_rows, + *args, + **expected_kwargs, + ) + + +def test_insert_rows_general_error(insert_rows_setup): + service, args, expected_kwargs = insert_rows_setup + throws_error_test( + RequestError, service.get_general_error_response(), insert_rows, *args, **expected_kwargs + ) + + +@pytest.fixture +def execute_sql_setup(): + service = MockExecuteSQL() + sql = "select * from " + schema + "." + query + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"sql": waf_encode(sql), "schemaName": schema}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema, sql] + return service, args, expected_kwargs + + +def test_execute_sql_success(execute_sql_setup): + service, args, expected_kwargs = execute_sql_setup + success_test(service.get_successful_response(), execute_sql, True, *args, **expected_kwargs) + + +def test_execute_sql_unauthorized(execute_sql_setup): + service, args, expected_kwargs = execute_sql_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + execute_sql, + *args, + **expected_kwargs, + ) + + +def test_execute_sql_query_not_found(execute_sql_setup): + service, args, expected_kwargs = execute_sql_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + execute_sql, + *args, + **expected_kwargs, + ) + + +def test_execute_sql_server_not_found(execute_sql_setup): + service, args, expected_kwargs = execute_sql_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + execute_sql, + *args, + **expected_kwargs, + ) + + +def test_execute_sql_general_error(execute_sql_setup): + service, args, expected_kwargs = execute_sql_setup + throws_error_test( + RequestError, service.get_general_error_response(), execute_sql, *args, **expected_kwargs + ) + + +@pytest.fixture +def select_rows_setup(): + service = MockSelectRows() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"schemaName": schema, "query.queryName": query, "query.maxRows": -1}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema, query] + return service, args, expected_kwargs + + +def test_select_rows_success(select_rows_setup): + service, args, expected_kwargs = select_rows_setup + success_test(service.get_successful_response(), select_rows, True, *args, **expected_kwargs) + + +def test_select_rows_query_filter(select_rows_setup): + service, args, expected_kwargs = select_rows_setup + view_name = None + filter_array = [ + QueryFilter("Field1", "value", "eq"), + QueryFilter("Field2", "value1", "contains"), + QueryFilter("Field2", "value2", "contains"), + ] + test_args = list(args) + [view_name, filter_array] + # Expected query field values in post request body + query_field = { + "query.Field1~eq": ["value"], + "query.Field2~contains": ["value1", "value2"], + } + # Update post request body with expected query field values + expected_kwargs["data"].update(query_field) + + success_test( + service.get_successful_response(), + select_rows, + True, + *test_args, + **expected_kwargs, + ) + + +def test_select_rows_unauthorized(select_rows_setup): + service, args, expected_kwargs = select_rows_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + select_rows, + *args, + **expected_kwargs, + ) + + +def test_select_rows_query_not_found(select_rows_setup): + service, args, expected_kwargs = select_rows_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + select_rows, + *args, + **expected_kwargs, + ) + + +def test_select_rows_server_not_found(select_rows_setup): + service, args, expected_kwargs = select_rows_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + select_rows, + *args, + **expected_kwargs, + ) + + +def test_select_rows_general_error(select_rows_setup): + service, args, expected_kwargs = select_rows_setup + throws_error_test( + RequestError, service.get_general_error_response(), select_rows, *args, **expected_kwargs + ) + + +@pytest.fixture +def save_rows_setup(): + service = MockSaveRows() + commands = [ + { + "command": "insert", + "schema_name": "exp", + "query_name": "materials", + "rows": [{"name": "New Sample 1"}], } - # Update post request body with expected query field values - self.expected_kwargs["data"].update(query_field) - - success_test( - test, - self.service.get_successful_response(), - select_rows, - True, - *args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - select_rows, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - select_rows, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - select_rows, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - select_rows, - *self.args, - **self.expected_kwargs - ) - - -class TestSaveRows(unittest.TestCase): - def setUp(self): - self.service = MockSaveRows() - commands = [ + ] + + expected_payload = { + "commands": [ { "command": "insert", - "schema_name": "exp", - "query_name": "materials", + "schemaName": "exp", + "queryName": "materials", "rows": [{"name": "New Sample 1"}], } ] + } + + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": json.dumps(expected_payload, sort_keys=True), + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + + args = [mock_server_context(service), commands] + return service, args, expected_kwargs + + +def test_save_rows_success(save_rows_setup): + service, args, expected_kwargs = save_rows_setup + success_test(service.get_successful_response(), save_rows, True, *args, **expected_kwargs) + + +def test_save_rows_unauthorized(save_rows_setup): + service, args, expected_kwargs = save_rows_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + save_rows, + *args, + **expected_kwargs, + ) - expected_payload = { - "commands": [ - { - "command": "insert", - "schemaName": "exp", - "queryName": "materials", - "rows": [{"name": "New Sample 1"}], - } - ] - } - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": json.dumps(expected_payload, sort_keys=True), - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, +def test_save_rows_query_not_found(save_rows_setup): + service, args, expected_kwargs = save_rows_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + save_rows, + *args, + **expected_kwargs, + ) + + +def test_save_rows_server_not_found(save_rows_setup): + service, args, expected_kwargs = save_rows_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + save_rows, + *args, + **expected_kwargs, + ) + + +def test_save_rows_general_error(save_rows_setup): + service, args, expected_kwargs = save_rows_setup + throws_error_test( + RequestError, service.get_general_error_response(), save_rows, *args, **expected_kwargs + ) + + +def test_save_rows_with_optional_command_fields(save_rows_setup): + service, _, _ = save_rows_setup + """Test save_rows with all optional fields populated in command""" + commands = [ + { + "command": "update", + "schema_name": "exp", + "query_name": "materials", + "rows": [{"rowId": 1, "name": "Updated Sample"}], + "audit_behavior": "DETAILED", + "audit_user_comment": "Test update comment", + "container_path": "/custom/path", + "extra_context": {"custom": "data"}, + "skip_reselect_rows": True, } + ] - self.args = [mock_server_context(self.service), commands] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - save_rows, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - save_rows, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - save_rows, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - save_rows, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - save_rows, - *self.args, - **self.expected_kwargs - ) - - def test_with_optional_command_fields(self): - """Test save_rows with all optional fields populated in command""" - commands = [ + expected_payload = { + "commands": [ { "command": "update", - "schema_name": "exp", - "query_name": "materials", + "schemaName": "exp", + "queryName": "materials", "rows": [{"rowId": 1, "name": "Updated Sample"}], - "audit_behavior": "DETAILED", - "audit_user_comment": "Test update comment", - "container_path": "/custom/path", - "extra_context": {"custom": "data"}, - "skip_reselect_rows": True, + "auditBehavior": "DETAILED", + "auditUserComment": "Test update comment", + "containerPath": "/custom/path", + "extraContext": {"custom": "data"}, + "skipReselectRows": True, } ] - - expected_payload = { - "commands": [ - { - "command": "update", - "schemaName": "exp", - "queryName": "materials", - "rows": [{"rowId": 1, "name": "Updated Sample"}], - "auditBehavior": "DETAILED", - "auditUserComment": "Test update comment", - "containerPath": "/custom/path", - "extraContext": {"custom": "data"}, - "skipReselectRows": True, - } - ] - } - - expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": json.dumps(expected_payload, sort_keys=True), - "headers": {"Content-Type": "application/json"}, - "timeout": 300, - "allow_redirects": False, + } + + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": json.dumps(expected_payload, sort_keys=True), + "headers": {"Content-Type": "application/json"}, + "timeout": 300, + "allow_redirects": False, + } + + args = [mock_server_context(service), commands] + + success_test(service.get_successful_response(), save_rows, True, *args, **expected_kwargs) + + +def test_save_rows_with_optional_parameters(save_rows_setup): + service, _, _ = save_rows_setup + api_version = 18.21 + commands = [ + { + "command": "delete", + "schema_name": "exp", + "query_name": "materials", + "rows": [{"rowId": 1}], } - - args = [mock_server_context(self.service), commands] - - success_test( - self, self.service.get_successful_response(), save_rows, True, *args, **expected_kwargs - ) - - def test_with_optional_parameters(self): - api_version = 18.21 - commands = [ + ] + extra_context = {"global": "context"} + transacted = False + timeout = 600 + validate_only = True + + expected_payload = { + "commands": [ { "command": "delete", - "schema_name": "exp", - "query_name": "materials", + "schemaName": "exp", + "queryName": "materials", "rows": [{"rowId": 1}], } - ] - extra_context = {"global": "context"} - transacted = False - timeout = 600 - validate_only = True - - expected_payload = { - "commands": [ - { - "command": "delete", - "schemaName": "exp", - "queryName": "materials", - "rows": [{"rowId": 1}], - } - ], - "apiVersion": api_version, - "extraContext": extra_context, - "transacted": transacted, - "validateOnly": True, - } - - expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": json.dumps(expected_payload, sort_keys=True), - "headers": {"Content-Type": "application/json"}, - "timeout": timeout, - "allow_redirects": False, + ], + "apiVersion": api_version, + "extraContext": extra_context, + "transacted": transacted, + "validateOnly": True, + } + + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": json.dumps(expected_payload, sort_keys=True), + "headers": {"Content-Type": "application/json"}, + "timeout": timeout, + "allow_redirects": False, + } + + args = [ + mock_server_context(service), + commands, + api_version, + None, # container_path + extra_context, + timeout, + transacted, + validate_only, + ] + + success_test(service.get_successful_response(), save_rows, True, *args, **expected_kwargs) + + +@pytest.fixture +def get_queries_setup(): + service = MockGetQueries() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"schemaName": schema}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), schema] + return service, args, expected_kwargs + + +def test_get_queries_success(get_queries_setup): + service, args, expected_kwargs = get_queries_setup + success_test(service.get_successful_response(), get_queries, True, *args, **expected_kwargs) + + +def test_get_queries_camel_case(get_queries_setup): + service, args, expected_kwargs = get_queries_setup + options = { + "include_columns": True, + "include_system_queries": False, + "include_title": True, + "include_user_queries": False, + "include_view_data_url": True, + "query_detail_columns": False, + } + expected_kwargs["data"].update( + { + "includeColumns": True, + "includeSystemQueries": False, + "includeTitle": True, + "includeUserQueries": False, + "includeViewDataUrl": True, + "queryDetailColumns": False, } - - args = [ - mock_server_context(self.service), - commands, - api_version, - None, # container_path - extra_context, - timeout, - transacted, - validate_only, - ] - - success_test( - self, self.service.get_successful_response(), save_rows, True, *args, **expected_kwargs - ) - - -def suite(): - load_tests = unittest.TestLoader().loadTestsFromTestCase - return unittest.TestSuite( - [ - load_tests(TestDeleteRows), - load_tests(TestUpdateRows), - load_tests(TestInsertRows), - load_tests(TestExecuteSQL), - load_tests(TestSelectRows), - load_tests(TestSaveRows), - ] ) - -if __name__ == "__main__": - unittest.main() + # We use a lambda below because get_queries uses kwargs, but success_test doesn't have a way to pass kwargs + success_test( + service.get_successful_response(), + lambda *a: get_queries(*a, **options), + True, + *args, + **expected_kwargs, + ) diff --git a/test/unit/test_security.py b/test/unit/test_security.py index 4c1e3ee..7860213 100644 --- a/test/unit/test_security.py +++ b/test/unit/test_security.py @@ -13,9 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. # -import unittest - -import unittest.mock as mock +import pytest from labkey.security import ( create_user, @@ -52,854 +50,729 @@ class MockUserController(MockLabKey): use_ssl = False -class TestCreateUser(unittest.TestCase): - __email = "pyTest@labkey.com" +@pytest.fixture +def create_user_setup(): + email = "pyTest@labkey.com" class MockCreateUser(MockSecurityController): api = "createNewUser.api" - def setUp(self): - self.service = self.MockCreateUser() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"email": TestCreateUser.__email, "sendEmail": False}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), self.__email] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - create_user, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - create_user, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - create_user, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - create_user, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - create_user, - *self.args, - **self.expected_kwargs - ) - - -class TestResetPassword(unittest.TestCase): - __email = "pyTest@labkey.com" + service = MockCreateUser() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"email": email, "sendEmail": False}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), email] + return service, args, expected_kwargs + + +def test_create_user_success(create_user_setup): + service, args, expected_kwargs = create_user_setup + success_test(service.get_successful_response(), create_user, True, *args, **expected_kwargs) + + +def test_create_user_unauthorized(create_user_setup): + service, args, expected_kwargs = create_user_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + create_user, + *args, + **expected_kwargs, + ) + + +def test_create_user_query_not_found(create_user_setup): + service, args, expected_kwargs = create_user_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + create_user, + *args, + **expected_kwargs, + ) + + +def test_create_user_server_not_found(create_user_setup): + service, args, expected_kwargs = create_user_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + create_user, + *args, + **expected_kwargs, + ) + + +def test_create_user_general_error(create_user_setup): + service, args, expected_kwargs = create_user_setup + throws_error_test( + RequestError, service.get_general_error_response(), create_user, *args, **expected_kwargs + ) + + +@pytest.fixture +def reset_password_setup(): + email = "pyTest@labkey.com" class MockResetPassword(MockSecurityController): api = "adminRotatePassword.api" - def setUp(self): - self.service = self.MockResetPassword() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"email": TestResetPassword.__email}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), self.__email] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - reset_password, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - reset_password, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - reset_password, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - reset_password, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - reset_password, - *self.args, - **self.expected_kwargs - ) - - -class TestActivateUsers(unittest.TestCase): - __user_id = [123] + service = MockResetPassword() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"email": email}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), email] + return service, args, expected_kwargs + + +def test_reset_password_success(reset_password_setup): + service, args, expected_kwargs = reset_password_setup + success_test(service.get_successful_response(), reset_password, True, *args, **expected_kwargs) + + +def test_reset_password_unauthorized(reset_password_setup): + service, args, expected_kwargs = reset_password_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + reset_password, + *args, + **expected_kwargs, + ) + + +def test_reset_password_query_not_found(reset_password_setup): + service, args, expected_kwargs = reset_password_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + reset_password, + *args, + **expected_kwargs, + ) + + +def test_reset_password_server_not_found(reset_password_setup): + service, args, expected_kwargs = reset_password_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + reset_password, + *args, + **expected_kwargs, + ) + + +def test_reset_password_general_error(reset_password_setup): + service, args, expected_kwargs = reset_password_setup + throws_error_test( + RequestError, service.get_general_error_response(), reset_password, *args, **expected_kwargs + ) + + +@pytest.fixture +def activate_users_setup(): + user_ids = [123] class MockActivateUser(MockUserController): api = "activateUsers.api" - def setUp(self): - self.service = self.MockActivateUser() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"userId": [123]}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), self.__user_id] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - activate_users, - True, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - activate_users, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - activate_users, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - activate_users, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - activate_users, - *self.args, - **self.expected_kwargs - ) - - -class TestDeactivateUsers(unittest.TestCase): - __user_id = [123] + service = MockActivateUser() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"userId": user_ids}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), user_ids] + return service, args, expected_kwargs + + +def test_activate_users_success(activate_users_setup): + service, args, expected_kwargs = activate_users_setup + success_test(service.get_successful_response(), activate_users, True, *args, **expected_kwargs) + + +def test_activate_users_unauthorized(activate_users_setup): + service, args, expected_kwargs = activate_users_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + activate_users, + *args, + **expected_kwargs, + ) + + +def test_activate_users_query_not_found(activate_users_setup): + service, args, expected_kwargs = activate_users_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + activate_users, + *args, + **expected_kwargs, + ) + + +def test_activate_users_server_not_found(activate_users_setup): + service, args, expected_kwargs = activate_users_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + activate_users, + *args, + **expected_kwargs, + ) + + +def test_activate_users_general_error(activate_users_setup): + service, args, expected_kwargs = activate_users_setup + throws_error_test( + RequestError, service.get_general_error_response(), activate_users, *args, **expected_kwargs + ) + + +@pytest.fixture +def deactivate_users_setup(): + user_ids = [123] class MockDeactivateUser(MockUserController): api = "deactivateUsers.view" - def setUp(self): - self.service = self.MockDeactivateUser() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"userId": [123]}, - "headers": None, - "timeout": 300, - "allow_redirects": True, - } - - self.args = [mock_server_context(self.service), self.__user_id] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - deactivate_users, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - deactivate_users, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - deactivate_users, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - deactivate_users, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - deactivate_users, - *self.args, - **self.expected_kwargs - ) - - -class TestDeleteUsers(unittest.TestCase): - __user_id = [123] + service = MockDeactivateUser() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"userId": user_ids}, + "headers": None, + "timeout": 300, + "allow_redirects": True, + } + args = [mock_server_context(service), user_ids] + return service, args, expected_kwargs + + +def test_deactivate_users_success(deactivate_users_setup): + service, args, expected_kwargs = deactivate_users_setup + success_test( + service.get_successful_response(), deactivate_users, False, *args, **expected_kwargs + ) + + +def test_deactivate_users_unauthorized(deactivate_users_setup): + service, args, expected_kwargs = deactivate_users_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + deactivate_users, + *args, + **expected_kwargs, + ) + + +def test_deactivate_users_query_not_found(deactivate_users_setup): + service, args, expected_kwargs = deactivate_users_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + deactivate_users, + *args, + **expected_kwargs, + ) + + +def test_deactivate_users_server_not_found(deactivate_users_setup): + service, args, expected_kwargs = deactivate_users_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + deactivate_users, + *args, + **expected_kwargs, + ) + + +def test_deactivate_users_general_error(deactivate_users_setup): + service, args, expected_kwargs = deactivate_users_setup + throws_error_test( + RequestError, service.get_general_error_response(), deactivate_users, *args, **expected_kwargs + ) + + +@pytest.fixture +def delete_users_setup(): + user_ids = [123] class MockDeleteUser(MockUserController): api = "deleteUsers.view" - def setUp(self): - self.service = self.MockDeleteUser() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"userId": [123]}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), self.__user_id] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - delete_users, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - delete_users, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - delete_users, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - delete_users, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - delete_users, - *self.args, - **self.expected_kwargs - ) - - -class TestAddToGroup(unittest.TestCase): - __user_id = 321 - __group_id = 123 + service = MockDeleteUser() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"userId": user_ids}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), user_ids] + return service, args, expected_kwargs + + +def test_delete_users_success(delete_users_setup): + service, args, expected_kwargs = delete_users_setup + success_test(service.get_successful_response(), delete_users, False, *args, **expected_kwargs) + + +def test_delete_users_unauthorized(delete_users_setup): + service, args, expected_kwargs = delete_users_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + delete_users, + *args, + **expected_kwargs, + ) + + +def test_delete_users_query_not_found(delete_users_setup): + service, args, expected_kwargs = delete_users_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + delete_users, + *args, + **expected_kwargs, + ) + + +def test_delete_users_server_not_found(delete_users_setup): + service, args, expected_kwargs = delete_users_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + delete_users, + *args, + **expected_kwargs, + ) + + +def test_delete_users_general_error(delete_users_setup): + service, args, expected_kwargs = delete_users_setup + throws_error_test( + RequestError, service.get_general_error_response(), delete_users, *args, **expected_kwargs + ) + + +@pytest.fixture +def add_to_group_setup(): + user_id = 321 + group_id = 123 class MockAddGroupMember(MockSecurityController): api = "addGroupMember.api" - def setUp(self): - self.service = self.MockAddGroupMember() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"groupId": 123, "principalIds": [321]}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), self.__user_id, self.__group_id] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - add_to_group, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - add_to_group, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - add_to_group, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - add_to_group, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - add_to_group, - *self.args, - **self.expected_kwargs - ) - - -class TestRemoveFromGroup(unittest.TestCase): - __user_id = 321 - __group_id = 123 + service = MockAddGroupMember() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"groupId": 123, "principalIds": [321]}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), user_id, group_id] + return service, args, expected_kwargs + + +def test_add_to_group_success(add_to_group_setup): + service, args, expected_kwargs = add_to_group_setup + success_test(service.get_successful_response(), add_to_group, False, *args, **expected_kwargs) + + +def test_add_to_group_unauthorized(add_to_group_setup): + service, args, expected_kwargs = add_to_group_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + add_to_group, + *args, + **expected_kwargs, + ) + + +def test_add_to_group_query_not_found(add_to_group_setup): + service, args, expected_kwargs = add_to_group_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + add_to_group, + *args, + **expected_kwargs, + ) + + +def test_add_to_group_server_not_found(add_to_group_setup): + service, args, expected_kwargs = add_to_group_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + add_to_group, + *args, + **expected_kwargs, + ) + + +def test_add_to_group_general_error(add_to_group_setup): + service, args, expected_kwargs = add_to_group_setup + throws_error_test( + RequestError, service.get_general_error_response(), add_to_group, *args, **expected_kwargs + ) + + +@pytest.fixture +def remove_from_group_setup(): + user_id = 321 + group_id = 123 class MockRemoveGroupMember(MockSecurityController): api = "removeGroupMember.api" - def setUp(self): - self.service = self.MockRemoveGroupMember() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"groupId": 123, "principalIds": [321]}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), self.__user_id, self.__group_id] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - remove_from_group, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - remove_from_group, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - remove_from_group, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - remove_from_group, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - remove_from_group, - *self.args, - **self.expected_kwargs - ) - - -class TestRemoveFromRole(unittest.TestCase): - __user_id = 321 - __email = "pyTest@labkey.com" - __role = {"uniqueName": "TestRole"} + service = MockRemoveGroupMember() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"groupId": 123, "principalIds": [321]}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), user_id, group_id] + return service, args, expected_kwargs + + +def test_remove_from_group_success(remove_from_group_setup): + service, args, expected_kwargs = remove_from_group_setup + success_test( + service.get_successful_response(), remove_from_group, False, *args, **expected_kwargs + ) + + +def test_remove_from_group_unauthorized(remove_from_group_setup): + service, args, expected_kwargs = remove_from_group_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + remove_from_group, + *args, + **expected_kwargs, + ) + + +def test_remove_from_group_query_not_found(remove_from_group_setup): + service, args, expected_kwargs = remove_from_group_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + remove_from_group, + *args, + **expected_kwargs, + ) + + +def test_remove_from_group_server_not_found(remove_from_group_setup): + service, args, expected_kwargs = remove_from_group_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + remove_from_group, + *args, + **expected_kwargs, + ) + + +def test_remove_from_group_general_error(remove_from_group_setup): + service, args, expected_kwargs = remove_from_group_setup + throws_error_test( + RequestError, + service.get_general_error_response(), + remove_from_group, + *args, + **expected_kwargs, + ) + + +@pytest.fixture +def remove_from_role_setup(): + user_id = 321 + email = "pyTest@labkey.com" + role = {"uniqueName": "TestRole"} class MockRemoveRole(MockSecurityController): api = "removeAssignment.api" - def setUp(self): - self.service = self.MockRemoveRole() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": { - "roleClassName": "TestRole", - "principalId": 321, - "email": "pyTest@labkey.com", - }, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [ - mock_server_context(self.service), - self.__role, - self.__user_id, - self.__email, - ] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - remove_from_role, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - remove_from_role, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - remove_from_role, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - remove_from_role, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - remove_from_role, - *self.args, - **self.expected_kwargs - ) - - -class TestAddToRole(unittest.TestCase): - __user_id = 321 - __email = "pyTest@labkey.com" - __role = {"uniqueName": "TestRole"} + service = MockRemoveRole() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": { + "roleClassName": "TestRole", + "principalId": 321, + "email": "pyTest@labkey.com", + }, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), role, user_id, email] + return service, args, expected_kwargs + + +def test_remove_from_role_success(remove_from_role_setup): + service, args, expected_kwargs = remove_from_role_setup + success_test( + service.get_successful_response(), remove_from_role, False, *args, **expected_kwargs + ) + + +def test_remove_from_role_unauthorized(remove_from_role_setup): + service, args, expected_kwargs = remove_from_role_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + remove_from_role, + *args, + **expected_kwargs, + ) + + +def test_remove_from_role_query_not_found(remove_from_role_setup): + service, args, expected_kwargs = remove_from_role_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + remove_from_role, + *args, + **expected_kwargs, + ) + + +def test_remove_from_role_server_not_found(remove_from_role_setup): + service, args, expected_kwargs = remove_from_role_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + remove_from_role, + *args, + **expected_kwargs, + ) + + +def test_remove_from_role_general_error(remove_from_role_setup): + service, args, expected_kwargs = remove_from_role_setup + throws_error_test( + RequestError, + service.get_general_error_response(), + remove_from_role, + *args, + **expected_kwargs, + ) + + +@pytest.fixture +def add_to_role_setup(): + user_id = 321 + email = "pyTest@labkey.com" + role = {"uniqueName": "TestRole"} class MockAddRole(MockSecurityController): api = "addAssignment.api" - def setUp(self): - self.service = self.MockAddRole() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": { - "roleClassName": "TestRole", - "principalId": 321, - "email": "pyTest@labkey.com", - }, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [ - mock_server_context(self.service), - self.__role, - self.__user_id, - self.__email, - ] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - add_to_role, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - add_to_role, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - add_to_role, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - add_to_role, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - add_to_role, - *self.args, - **self.expected_kwargs - ) - - -class TestGetRoles(unittest.TestCase): + service = MockAddRole() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": { + "roleClassName": "TestRole", + "principalId": 321, + "email": "pyTest@labkey.com", + }, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), role, user_id, email] + return service, args, expected_kwargs + + +def test_add_to_role_success(add_to_role_setup): + service, args, expected_kwargs = add_to_role_setup + success_test(service.get_successful_response(), add_to_role, False, *args, **expected_kwargs) + + +def test_add_to_role_unauthorized(add_to_role_setup): + service, args, expected_kwargs = add_to_role_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + add_to_role, + *args, + **expected_kwargs, + ) + + +def test_add_to_role_query_not_found(add_to_role_setup): + service, args, expected_kwargs = add_to_role_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + add_to_role, + *args, + **expected_kwargs, + ) + + +def test_add_to_role_server_not_found(add_to_role_setup): + service, args, expected_kwargs = add_to_role_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + add_to_role, + *args, + **expected_kwargs, + ) + + +def test_add_to_role_general_error(add_to_role_setup): + service, args, expected_kwargs = add_to_role_setup + throws_error_test( + RequestError, service.get_general_error_response(), add_to_role, *args, **expected_kwargs + ) + + +@pytest.fixture +def get_roles_setup(): class MockGetRoles(MockSecurityController): api = "getRoles.api" - def setUp(self): - self.service = self.MockGetRoles() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": None, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service)] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - get_roles, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - get_roles, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - get_roles, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - get_roles, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - get_roles, - *self.args, - **self.expected_kwargs - ) - - -class TestListGroups(unittest.TestCase): + service = MockGetRoles() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": None, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service)] + return service, args, expected_kwargs + + +def test_get_roles_success(get_roles_setup): + service, args, expected_kwargs = get_roles_setup + success_test(service.get_successful_response(), get_roles, False, *args, **expected_kwargs) + + +def test_get_roles_unauthorized(get_roles_setup): + service, args, expected_kwargs = get_roles_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + get_roles, + *args, + **expected_kwargs, + ) + + +def test_get_roles_query_not_found(get_roles_setup): + service, args, expected_kwargs = get_roles_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + get_roles, + *args, + **expected_kwargs, + ) + + +def test_get_roles_server_not_found(get_roles_setup): + service, args, expected_kwargs = get_roles_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + get_roles, + *args, + **expected_kwargs, + ) + + +def test_get_roles_general_error(get_roles_setup): + service, args, expected_kwargs = get_roles_setup + throws_error_test( + RequestError, service.get_general_error_response(), get_roles, *args, **expected_kwargs + ) + + +@pytest.fixture +def list_groups_setup(): class MockListGroups(MockSecurityController): api = "listProjectGroups.api" - def setUp(self): - self.service = self.MockListGroups() - self.expected_kwargs = { - "expected_args": [self.service.get_server_url()], - "data": {"includeSiteGroups": True}, - "headers": None, - "timeout": 300, - "allow_redirects": False, - } - - self.args = [mock_server_context(self.service), True] - - def test_success(self): - test = self - success_test( - test, - self.service.get_successful_response(), - list_groups, - False, - *self.args, - **self.expected_kwargs - ) - - def test_unauthorized(self): - test = self - throws_error_test( - test, - RequestAuthorizationError, - self.service.get_unauthorized_response(), - list_groups, - *self.args, - **self.expected_kwargs - ) - - def test_query_not_found(self): - test = self - throws_error_test( - test, - QueryNotFoundError, - self.service.get_query_not_found_response(), - list_groups, - *self.args, - **self.expected_kwargs - ) - - def test_server_not_found(self): - test = self - throws_error_test( - test, - ServerNotFoundError, - self.service.get_server_not_found_response(), - list_groups, - *self.args, - **self.expected_kwargs - ) - - def test_general_error(self): - test = self - throws_error_test( - test, - RequestError, - self.service.get_general_error_response(), - list_groups, - *self.args, - **self.expected_kwargs - ) - - -def suite(): - load_tests = unittest.TestLoader().loadTestsFromTestCase - return unittest.TestSuite( - [ - load_tests(TestCreateUser), - load_tests(TestActivateUsers), - load_tests(TestDeactivateUsers), - load_tests(TestDeleteUsers), - load_tests(TestRemoveFromGroup), - load_tests(TestAddToGroup), - load_tests(TestRemoveFromRole), - ] - ) - - -if __name__ == "__main__": - unittest.main() + service = MockListGroups() + expected_kwargs = { + "expected_args": [service.get_server_url()], + "data": {"includeSiteGroups": True}, + "headers": None, + "timeout": 300, + "allow_redirects": False, + } + args = [mock_server_context(service), True] + return service, args, expected_kwargs + + +def test_list_groups_success(list_groups_setup): + service, args, expected_kwargs = list_groups_setup + success_test(service.get_successful_response(), list_groups, False, *args, **expected_kwargs) + + +def test_list_groups_unauthorized(list_groups_setup): + service, args, expected_kwargs = list_groups_setup + throws_error_test( + RequestAuthorizationError, + service.get_unauthorized_response(), + list_groups, + *args, + **expected_kwargs, + ) + + +def test_list_groups_query_not_found(list_groups_setup): + service, args, expected_kwargs = list_groups_setup + throws_error_test( + QueryNotFoundError, + service.get_query_not_found_response(), + list_groups, + *args, + **expected_kwargs, + ) + + +def test_list_groups_server_not_found(list_groups_setup): + service, args, expected_kwargs = list_groups_setup + throws_error_test( + ServerNotFoundError, + service.get_server_not_found_response(), + list_groups, + *args, + **expected_kwargs, + ) + + +def test_list_groups_general_error(list_groups_setup): + service, args, expected_kwargs = list_groups_setup + throws_error_test( + RequestError, service.get_general_error_response(), list_groups, *args, **expected_kwargs + ) diff --git a/test/unit/test_utils.py b/test/unit/test_utils.py index 4648771..9f908a0 100644 --- a/test/unit/test_utils.py +++ b/test/unit/test_utils.py @@ -1,4 +1,4 @@ -from labkey.utils import btoa, encode_uri_component, waf_encode +from labkey.utils import btoa, encode_uri_component, waf_encode, snake_to_camel, transform_options def test_btoa(): @@ -30,3 +30,64 @@ def test_waf_encode(): waf_encode("><&/%' \"1äöüÅ") == prefix + "JTNFJTNDJTI2JTJGJTI1JyUyMCUyMjElQzMlQTQlQzMlQjYlQzMlQkMlQzMlODU=" ) + + +def test_snake_to_camel(): + assert snake_to_camel("snake_case") == "snakeCase" + assert snake_to_camel("multiple_word_snake_case") == "multipleWordSnakeCase" + assert snake_to_camel("alreadyCamelCase") == "alreadyCamelCase" + assert snake_to_camel("single") == "single" + assert snake_to_camel("UPPER_SNAKE_CASE") == "upperSnakeCase" + assert snake_to_camel("_leading_underscore") == "leadingUnderscore" + assert snake_to_camel("trailing_underscore_") == "trailingUnderscore" + assert snake_to_camel("multiple__underscores") == "multipleUnderscores" + assert snake_to_camel("") == "" + + +def test_transform_options(): + options = { + "include_columns": True, + "include_system_queries": False, + "include_title": False, + "include_user_queries": False, + "include_view_data_url": True, + "query_detail_columns": False, + } + expected_keys = [ + "include_columns", + "include_system_queries", + "include_user_queries", + "include_view_data_url", + ] + transformed_options = transform_options(options, expected_keys) + assert transformed_options == { + "includeColumns": True, + "includeSystemQueries": False, + "includeUserQueries": False, + "includeViewDataUrl": True, + } + assert transformed_options["includeColumns"] is True + + # Empty options + assert transform_options({}, ["any_key"]) == {} + + # Empty expected_keys + assert transform_options({"any_key": 1}, []) == {} + + # Filtering behavior: only keys in expected_keys are kept and transformed + options = {"keep_me": 1, "ignore_me": 2} + expected = ["keep_me"] + result = transform_options(options, expected) + assert result == {"keepMe": 1} + + # Keys in expected_keys but not in options are ignored + options = {"present_key": "I am present!"} + expected = ["present_key", "absent_key"] + result = transform_options(options, expected) + assert result == {"presentKey": "I am present!"} + + # Non-snake_case keys that are in expected_keys + options = {"alreadyCamel": 1, "simple": 2, "UPPER_SNAKE": 3} + expected = ["alreadyCamel", "simple", "UPPER_SNAKE"] + result = transform_options(options, expected) + assert result == {"alreadyCamel": 1, "simple": 2, "upperSnake": 3} diff --git a/test/unit/utilities.py b/test/unit/utilities.py index 5ac3d8f..2683c98 100644 --- a/test/unit/utilities.py +++ b/test/unit/utilities.py @@ -13,6 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import pytest import requests from labkey.server_context import ServerContext @@ -33,41 +34,37 @@ def mock_server_context(mock_action): ) -def success_test(test, expected_response, api_method, compare_response, *args, **expected_kwargs): +def success_test(expected_response, api_method, compare_response, *args, **expected_kwargs): with mock.patch("labkey.server_context.requests.Session.post") as mock_post: mock_post.return_value = expected_response resp = api_method(*args) # validate response is as expected if compare_response: - test.assertEqual(resp, expected_response.text) + assert resp == expected_response.text # validate call is made as expected expected_args = expected_kwargs.pop("expected_args") mock_post.assert_called_once_with(*expected_args, **expected_kwargs) -def success_test_get( - test, expected_response, api_method, compare_response, *args, **expected_kwargs -): +def success_test_get(expected_response, api_method, compare_response, *args, **expected_kwargs): with mock.patch("labkey.server_context.requests.Session.get") as mock_get: mock_get.return_value = expected_response resp = api_method(*args) # validate response is as expected if compare_response: - test.assertEqual(resp, expected_response.text) + assert resp == expected_response.text # validate call is made as expected expected_args = expected_kwargs.pop("expected_args") mock_get.assert_called_once_with(*expected_args, **expected_kwargs) -def throws_error_test( - test, expected_error, expected_response, api_method, *args, **expected_kwargs -): +def throws_error_test(expected_error, expected_response, api_method, *args, **expected_kwargs): with mock.patch("labkey.server_context.requests.Session.post") as mock_post: - with test.assertRaises(expected_error): + with pytest.raises(expected_error): mock_post.return_value = expected_response api_method(*args) @@ -76,11 +73,9 @@ def throws_error_test( mock_post.assert_called_once_with(*expected_args, **expected_kwargs) -def throws_error_test_get( - test, expected_error, expected_response, api_method, *args, **expected_kwargs -): +def throws_error_test_get(expected_error, expected_response, api_method, *args, **expected_kwargs): with mock.patch("labkey.server_context.requests.Session.get") as mock_get: - with test.assertRaises(expected_error): + with pytest.raises(expected_error): mock_get.return_value = expected_response api_method(*args)