-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Move auth logic to Bfabric class (#79)
- Make managing the `BfabricAuth` object the responsibility of the `Bfabric` class. This allows us to implement the contextmanager, which will be useful for the REST-proxy server. - Notably this shouldn't change anything about code that already uses the new class. - Rename `BfabricConfig.with_overrides` to `BfabricConfig.copy_with` for clarity. - Some unit tests for the Bfabric class, I will add more as I go.
- Loading branch information
1 parent
76daee1
commit 2afcd0c
Showing
6 changed files
with
145 additions
and
71 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,71 +1,74 @@ | ||
| from typing import Union, List | ||
| from __future__ import annotations | ||
|
|
||
| import copy | ||
| from typing import Any | ||
|
|
||
| from suds.client import Client | ||
| from suds import MethodNotFound | ||
| from suds.client import Client | ||
| from suds.serviceproxy import ServiceProxy | ||
|
|
||
| from bfabric.bfabric_config import BfabricAuth | ||
| from bfabric.src.errors import BfabricRequestError | ||
|
|
||
|
|
||
| class EngineSUDS: | ||
| """B-Fabric API SUDS Engine""" | ||
|
|
||
| def __init__(self, login: str, password: str, base_url: str): | ||
| def __init__(self, base_url: str) -> None: | ||
| self.cl = {} | ||
| self.login = login | ||
| self.password = password | ||
| self.base_url = base_url | ||
|
|
||
| def _get_client(self, endpoint: str): | ||
| try: | ||
| if endpoint not in self.cl: | ||
| wsdl = "".join((self.base_url, '/', endpoint, "?wsdl")) | ||
| self.cl[endpoint] = Client(wsdl, cache=None) | ||
| return self.cl[endpoint] | ||
| except Exception as e: | ||
| print(e) | ||
| raise | ||
|
|
||
| def read(self, endpoint: str, obj: dict, page: int = 1, idonly: bool = False, | ||
| includedeletableupdateable: bool = False): | ||
| """ | ||
| A generic method which can connect to any endpoint, e.g., workunit, project, order, | ||
| externaljob, etc, and returns the object with the requested id. | ||
| obj is a python dictionary which contains all the attributes of the endpoint | ||
| for the "query". | ||
| def read( | ||
| self, | ||
| endpoint: str, | ||
| obj: dict[str, Any], | ||
| auth: BfabricAuth, | ||
| page: int = 1, | ||
| idonly: bool = False, | ||
| includedeletableupdateable: bool = False, | ||
| ): | ||
| """Reads the requested `obj` from `endpoint`. | ||
| :param endpoint: the endpoint to read, e.g. `workunit`, `project`, `order`, `externaljob`, etc. | ||
| :param obj: a python dictionary which contains all the attribute values that have to match | ||
| :param auth: the authentication handle of the user performing the request | ||
| :param page: the page number to read | ||
| :param idonly: whether to return only the ids of the objects | ||
| :param includedeletableupdateable: TODO | ||
| """ | ||
| query = copy.deepcopy(obj) | ||
| query['includedeletableupdateable'] = includedeletableupdateable | ||
|
|
||
| full_query = dict(login=self.login, page=page, password=self.password, query=query, | ||
| idonly=idonly) | ||
| query["includedeletableupdateable"] = includedeletableupdateable | ||
|
|
||
| client = self._get_client(endpoint) | ||
| return client.service.read(full_query) | ||
| full_query = dict(login=auth.login, page=page, password=auth.password, query=query, idonly=idonly) | ||
| service = self._get_suds_service(endpoint) | ||
| return service.read(full_query) | ||
|
|
||
| # TODO: How is client.service.readid different from client.service.read. Do we need this method? | ||
| def readid(self, endpoint: str, obj: dict, page: int = 1): | ||
| query = dict(login=self.login, page=page, password=self.password, query=obj) | ||
|
|
||
| client = self._get_client(endpoint) | ||
| return client.service.readid(query) | ||
|
|
||
| def save(self, endpoint: str, obj: dict): | ||
| query = {'login': self.login, 'password': self.password, endpoint: obj} | ||
|
|
||
| client = self._get_client(endpoint) | ||
| def readid(self, endpoint: str, obj: dict, auth: BfabricAuth, page: int = 1): | ||
| query = dict(login=auth.login, page=page, password=auth.password, query=obj) | ||
| service = self._get_suds_service(endpoint) | ||
| return service.readid(query) | ||
|
|
||
| def save(self, endpoint: str, obj: dict, auth: BfabricAuth): | ||
| query = {"login": auth.login, "password": auth.password, endpoint: obj} | ||
| service = self._get_suds_service(endpoint) | ||
| try: | ||
| res = client.service.save(query) | ||
| res = service.save(query) | ||
| except MethodNotFound as e: | ||
| raise BfabricRequestError(f"SUDS failed to find save method for the {endpoint} endpoint.") from e | ||
| return res | ||
|
|
||
| def delete(self, endpoint: str, id: Union[int, List]): | ||
| def delete(self, endpoint: str, id: int | list[int], auth: BfabricAuth): | ||
| if isinstance(id, list) and len(id) == 0: | ||
| print("Warning, attempted to delete an empty list, ignoring") | ||
| return [] | ||
|
|
||
| query = {'login': self.login, 'password': self.password, 'id': id} | ||
| query = {"login": auth.login, "password": auth.password, "id": id} | ||
| service = self._get_suds_service(endpoint) | ||
| return service.delete(query) | ||
|
|
||
| client = self._get_client(endpoint) | ||
| return client.service.delete(query) | ||
| def _get_suds_service(self, endpoint: str) -> ServiceProxy: | ||
| """Returns a SUDS service for the given endpoint. Reuses existing instances when possible.""" | ||
| if endpoint not in self.cl: | ||
| wsdl = "".join((self.base_url, "/", endpoint, "?wsdl")) | ||
| self.cl[endpoint] = Client(wsdl, cache=None) | ||
| return self.cl[endpoint].service |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| import unittest | ||
| from functools import cached_property | ||
| from unittest.mock import MagicMock | ||
|
|
||
| from bfabric import BfabricConfig | ||
| from bfabric.bfabric2 import BfabricAPIEngineType, Bfabric | ||
| from bfabric.src.engine_suds import EngineSUDS | ||
|
|
||
|
|
||
| class TestBfabric(unittest.TestCase): | ||
| def setUp(self): | ||
| self.mock_config = MagicMock(name="mock_config", spec=BfabricConfig) | ||
| self.mock_auth = None | ||
| self.mock_engine_type = BfabricAPIEngineType.SUDS | ||
| self.mock_engine = MagicMock(name="mock_engine", spec=EngineSUDS) | ||
|
|
||
| @cached_property | ||
| def mock_bfabric(self) -> Bfabric: | ||
| return Bfabric(config=self.mock_config, auth=self.mock_auth, engine=self.mock_engine_type) | ||
|
|
||
| def test_query_counter(self): | ||
| self.assertEqual(0, self.mock_bfabric.query_counter) | ||
|
|
||
| def test_config(self): | ||
| self.assertEqual(self.mock_config, self.mock_bfabric.config) | ||
|
|
||
| def test_auth_when_missing(self): | ||
| with self.assertRaises(ValueError) as error: | ||
| _ = self.mock_bfabric.auth | ||
| self.assertIn("Authentication not available", str(error.exception)) | ||
|
|
||
| def test_auth_when_provided(self): | ||
| self.mock_auth = MagicMock(name="mock_auth") | ||
| self.assertEqual(self.mock_auth, self.mock_bfabric.auth) | ||
|
|
||
| def test_with_auth(self): | ||
| mock_old_auth = MagicMock(name="mock_old_auth") | ||
| mock_new_auth = MagicMock(name="mock_new_auth") | ||
| self.mock_auth = mock_old_auth | ||
| with self.mock_bfabric.with_auth(mock_new_auth): | ||
| self.assertEqual(mock_new_auth, self.mock_bfabric.auth) | ||
| self.assertEqual(mock_old_auth, self.mock_bfabric.auth) | ||
|
|
||
| def test_with_auth_when_exception(self): | ||
| mock_old_auth = MagicMock(name="mock_old_auth") | ||
| mock_new_auth = MagicMock(name="mock_new_auth") | ||
| self.mock_auth = mock_old_auth | ||
| try: | ||
| with self.mock_bfabric.with_auth(mock_new_auth): | ||
| raise ValueError("Test exception") | ||
| except ValueError: | ||
| pass | ||
| self.assertEqual(mock_old_auth, self.mock_bfabric.auth) | ||
|
|
||
| # TODO further unit tests | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters