diff --git a/atst/eda_client.py b/atst/eda_client.py index 3251499e..2c9b65f5 100644 --- a/atst/eda_client.py +++ b/atst/eda_client.py @@ -1,3 +1,10 @@ +from csv import DictReader +import xml.etree.ElementTree as ET + +import requests +from requests.auth import HTTPBasicAuth + + class EDAClientBase(object): def list_contracts( self, @@ -97,8 +104,24 @@ class MockEDAClient(EDAClientBase): class EDAClient(EDAClientBase): - def __init__(self, base_url, user_name, user_role): - pass + def __init__(self, base_url, user_name, user_role, auth_name, auth_pass): + self.base_url = base_url + self.user_name = user_name + self.user_role = user_role + self.auth = HTTPBasicAuth(auth_name, auth_pass) + + def _make_url(self, method, **kwargs): + query_args = dict(kwargs) + query_string = "&".join( + ["{}={}".format(key, value) for key, value in query_args.items()] + ) + return "{base_url}/{method}?{query_string}".format( + base_url=self.base_url, method=method, query_string=query_string + ) + + def _get(self, method, **kwargs): + url = self._make_url(method, **kwargs) + return requests.get(url, auth=self.auth, verify="ssl/server-certs/eda.pem") def list_contracts( self, @@ -107,11 +130,37 @@ class EDAClient(EDAClientBase): cage_code=None, duns_number=None, ): - # TODO: Fetch the contracts CSV and transform them into dictionaries. - # https://docs.python.org/3/library/csv.html#csv.DictReader - raise NotImplementedError() + response = self._get( + "wawf_interface.returnContractList", + pContract=contract_number, + pDelivery_Order=delivery_order, + pCage_Code=cage_code, + pDuns_Number=duns_number, + pUserName=self.user_name, + pUser_Role=self.user_role, + ) + lines = response.text.replace("
", "").split("\n") + return list(DictReader(lines)) def get_contract(self, contract_number, status): - # TODO: Fetch the contract XML and transform it into a dictionary. - # https://docs.python.org/3.7/library/xml.etree.elementtree.html - raise NotImplementedError() + response = self._get( + "pds_contract_interface.get_xml_doc", + pContract=contract_number, + pStatus=status, + ) + if response.text.startswith("No data found"): + return None + return ET.fromstring(response.text) + + def get_clins(self, record_key, clins, cage_code="", duns_number=""): + response = self._get( + "wawf_interface.returnclinXML", + pCage_Code=cage_code, + pDuns_Number=duns_number, + pUserName=self.user_name, + pUser_Role=self.user_role, + pRecord_key=record_key, + pClins=clins, + ) + # TODO: Parse XML, similar to `get_contract` + return response