diff --git a/.gitignore b/.gitignore index 08cfdfd..aadc4fa 100644 --- a/.gitignore +++ b/.gitignore @@ -138,3 +138,4 @@ dmypy.json # Cython debug symbols cython_debug/ +src/southern_company_api/test.py diff --git a/src/southern_company_api/account.py b/src/southern_company_api/account.py index 884e455..15032d7 100644 --- a/src/southern_company_api/account.py +++ b/src/southern_company_api/account.py @@ -57,99 +57,101 @@ def __init__( self.hourly_data: typing.Dict[str, HourlyEnergyUsage] = {} self.daily_data: typing.Dict[str, DailyEnergyUsage] = {} self.session = session + self.service_point_number = None async def get_service_point_number(self, jwt: str) -> str: - async with self.session as session: - headers = {"Authorization": f"bearer {jwt}"} - # TODO: Is the /GPC for all customers or just GA power? - try: - async with session.get( - f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/" - f"getMPUBasicAccountInformation/{self.number}/GPC", - headers=headers, - ) as resp: + headers = { + "Authorization": f"bearer {jwt}", + "content-type": "application/json, text/plain, */*", + } + # TODO: Is the /GPC for all customers or just GA power? + try: + async with self.session.get( + f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/" + f"getMPUBasicAccountInformation/{self.number}/GPC", + headers=headers, + ) as resp: + try: + service_info = await resp.json() + except (ContentTypeError, json.JSONDecodeError) as err: try: - service_info = await resp.json() - except (ContentTypeError, json.JSONDecodeError) as err: - try: - error_text = await resp.text() - except aiohttp.ClientError: - error_text = err.msg - raise CantReachSouthernCompany( - f"Incorrect mimetype while trying to get service point number. {error_text}" - ) from err + error_text = await resp.text() + except aiohttp.ClientError: + error_text = err.msg + raise CantReachSouthernCompany( + f"Incorrect mimetype while trying to get service point number. error:{error_text} Response " + f"headers:{resp.headers} Your headers:{headers}" + ) from err - # TODO: Test with multiple accounts - return service_info["Data"]["meterAndServicePoints"][0][ - "servicePointNumber" - ] - except aiohttp.ClientConnectorError as err: - raise CantReachSouthernCompany("Failed to connect to api") from err + # TODO: Test with multiple accounts + self.service_point_number = service_info["Data"][ + "meterAndServicePoints" + ][0]["servicePointNumber"] + return service_info["Data"]["meterAndServicePoints"][0][ + "servicePointNumber" + ] + except aiohttp.ClientConnectorError as err: + raise CantReachSouthernCompany("Failed to connect to api") from err async def get_daily_data( self, start_date: datetime.datetime, end_date: datetime.datetime, jwt: str ) -> List[DailyEnergyUsage]: """Available 24 hours after""" """This is not really tested yet.""" - async with self.session as session: - headers = {"Authorization": f"bearer {jwt}"} - params = { - "accountNumber": self.number, - "startDate": start_date.strftime("%m/%d/%Y 12:00:00 AM"), - "endDate": end_date.strftime("%m/%d/%Y 11:59:59 PM"), - "OPCO": self.company.name, - "ServicePointNumber": await self.get_service_point_number(jwt), - "intervalBehavior": "Automatic", - } - async with session.get( - f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/" - f"MPUData/{self.number}/Daily", - headers=headers, - params=params, - ) as resp: - if resp.status != 200: - raise UsageDataFailure( - f"Failed to get daily data: {resp.status} {headers}" - ) - else: + headers = {"Authorization": f"bearer {jwt}"} + params = { + "accountNumber": self.number, + "startDate": start_date.strftime("%m/%d/%Y 12:00:00 AM"), + "endDate": end_date.strftime("%m/%d/%Y 11:59:59 PM"), + "OPCO": self.company.name, + "ServicePointNumber": self.service_point_number, + "intervalBehavior": "Automatic", + } + async with self.session.get( + f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/" + f"MPUData/{self.number}/Daily", + headers=headers, + params=params, + ) as resp: + if resp.status != 200: + raise UsageDataFailure( + f"Failed to get daily data: {resp.status} {headers}" + ) + else: + try: + response = await resp.json() + except (ContentTypeError, json.JSONDecodeError) as err: try: - response = await resp.json() - except (ContentTypeError, json.JSONDecodeError) as err: - try: - error_text = await resp.text() - except aiohttp.ClientErrors: - error_text = err.msg - raise CantReachSouthernCompany( - f"Incorrect mimetype while trying to get daily data. {error_text}" - ) from err - data = json.loads(response["Data"]["Data"]) - day_maps = {} - dates = [date for date in data["xAxis"]["labels"]] - high_temps = [ - temp["y"] for temp in data["series"]["highTemp"]["data"] - ] - low_temps = [ - temp["y"] for temp in data["series"]["lowTemp"]["data"] - ] - for i, date in enumerate(dates): - day_maps[date] = DailyEnergyUsage( - # TODO: Determine timezone - date=datetime.datetime.strptime(date, "%Y-%m-%dT%H:%M:%S"), - usage=-1, - cost=1, - low_temp=low_temps[i], - high_temp=high_temps[i], - ) - # TODO: Zip weekday and weekend to make it simpler. - for weekend_cost in data["series"]["weekdayCost"]["data"]: - day_maps[weekend_cost["name"]].cost = weekend_cost["y"] - for weekend_usage in data["series"]["weekdayUsage"]["data"]: - day_maps[weekend_usage["name"]].usage = weekend_usage["y"] - for weekday_cost in data["series"]["weekdayCost"]["data"]: - day_maps[weekday_cost["name"]].cost = weekday_cost["y"] - for weekday_usage in data["series"]["weekdayUsage"]["data"]: - day_maps[weekday_usage["name"]].usage = weekday_usage["y"] - return list(day_maps.values()) + error_text = await resp.text() + except aiohttp.ClientError: + error_text = err.msg + raise CantReachSouthernCompany( + f"Incorrect mimetype while trying to get daily data. {error_text}" + ) from err + data = json.loads(response["Data"]["Data"]) + day_maps = {} + dates = [date for date in data["xAxis"]["labels"]] + high_temps = [temp["y"] for temp in data["series"]["highTemp"]["data"]] + low_temps = [temp["y"] for temp in data["series"]["lowTemp"]["data"]] + for i, date in enumerate(dates): + day_maps[date] = DailyEnergyUsage( + # TODO: Determine timezone + date=datetime.datetime.strptime(date, "%Y-%m-%dT%H:%M:%S"), + usage=-1, + cost=1, + low_temp=low_temps[i], + high_temp=high_temps[i], + ) + # TODO: Zip weekday and weekend to make it simpler. + for weekend_cost in data["series"]["weekdayCost"]["data"]: + day_maps[weekend_cost["name"]].cost = weekend_cost["y"] + for weekend_usage in data["series"]["weekdayUsage"]["data"]: + day_maps[weekend_usage["name"]].usage = weekend_usage["y"] + for weekday_cost in data["series"]["weekdayCost"]["data"]: + day_maps[weekday_cost["name"]].cost = weekday_cost["y"] + for weekday_usage in data["series"]["weekdayUsage"]["data"]: + day_maps[weekday_usage["name"]].usage = weekday_usage["y"] + return list(day_maps.values()) async def get_hourly_data( self, start_date: datetime.datetime, end_date: datetime.datetime, jwt: str @@ -173,108 +175,102 @@ async def get_hourly_data( continue cur_date = cur_date + datetime.timedelta(days=35) return return_data - async with self.session as session: - # Needs to check if the data already exist in self.hourly_data to avoid making an unneeded call. - headers = {"Authorization": f"bearer {jwt}"} - params = { - "accountNumber": self.number, - "startDate": start_date.strftime("%m/%d/%Y %H:%M:%S %p"), - "endDate": end_date.strftime("%m/%d/%Y %H:%M:%S %p"), - "OPCO": self.company.name, - "ServicePointNumber": await self.get_service_point_number(jwt), - "intervalBehavior": "Automatic", - } - async with session.get( - f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/" - f"MPUData/{self.number}/Hourly", - headers=headers, - params=params, - ) as resp: - if resp.status != 200: - raise UsageDataFailure( - f"Failed to get hourly data: {resp.status} {headers}" - ) - else: + # Needs to check if the data already exist in self.hourly_data to avoid making an unneeded call. + headers = {"Authorization": f"bearer {jwt}"} + params = { + "accountNumber": self.number, + "startDate": start_date.strftime("%m/%d/%Y %H:%M:%S %p"), + "endDate": end_date.strftime("%m/%d/%Y %H:%M:%S %p"), + "OPCO": self.company.name, + "ServicePointNumber": self.service_point_number, + "intervalBehavior": "Automatic", + } + async with self.session.get( + f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/" + f"MPUData/{self.number}/Hourly", + headers=headers, + params=params, + ) as resp: + if resp.status != 200: + raise UsageDataFailure( + f"Failed to get hourly data: {resp.status} {headers}" + ) + else: + try: + data = await resp.json() + except (ContentTypeError, json.JSONDecodeError) as err: try: - data = await resp.json() - except (ContentTypeError, json.JSONDecodeError) as err: - try: - error_text = await resp.text() - except aiohttp.ClientError: - error_text = err.msg - raise CantReachSouthernCompany( - f"Incorrect mimetype while trying to get hourly data. {error_text}" - ) from err - if data["Data"]["Data"] is None: - raise UsageDataFailure("Received no data back for usage.") - data = json.loads(data["Data"]["Data"]) - return_dates = [] - for date in data["xAxis"]["labels"]: - # TODO: Determine timezone - parsed_date = datetime.datetime.strptime( - date, "%Y-%m-%dT%H:%M:%S" - ) - parsed_date = parsed_date.replace( - tzinfo=datetime.timezone( - datetime.timedelta(hours=-5), "EST" - ) - ) - self.hourly_data[date] = HourlyEnergyUsage( - time=parsed_date, usage=None, cost=None, temp=None - ) - return_dates.append(self.hourly_data[date]) - # costs and temps can be different lengths? - for cost in data["series"]["cost"]["data"]: - self.hourly_data[cost["name"]].cost = cost["y"] - for usage in data["series"]["usage"]["data"]: - self.hourly_data[usage["name"]].usage = usage["y"] - for temp in data["series"]["temp"]["data"]: - self.hourly_data[temp["name"]].temp = temp["y"] - return return_dates + error_text = await resp.text() + except aiohttp.ClientError: + error_text = err.msg + raise CantReachSouthernCompany( + f"Incorrect mimetype while trying to get hourly data. {error_text}" + ) from err + if data["Data"]["Data"] is None: + raise UsageDataFailure("Received no data back for usage.") + data = json.loads(data["Data"]["Data"]) + return_dates = [] + for date in data["xAxis"]["labels"]: + # TODO: Determine timezone + parsed_date = datetime.datetime.strptime(date, "%Y-%m-%dT%H:%M:%S") + parsed_date = parsed_date.replace( + tzinfo=datetime.timezone(datetime.timedelta(hours=-5), "EST") + ) + self.hourly_data[date] = HourlyEnergyUsage( + time=parsed_date, usage=None, cost=None, temp=None + ) + return_dates.append(self.hourly_data[date]) + # costs and temps can be different lengths? + for cost in data["series"]["cost"]["data"]: + self.hourly_data[cost["name"]].cost = cost["y"] + for usage in data["series"]["usage"]["data"]: + self.hourly_data[usage["name"]].usage = usage["y"] + for temp in data["series"]["temp"]["data"]: + self.hourly_data[temp["name"]].temp = temp["y"] + return return_dates async def get_month_data(self, jwt: str) -> MonthlyUsage: """Gets monthly data such as usage so far""" - async with self.session as session: - headers = {"Authorization": f"bearer {jwt}"} - today = datetime.datetime.now() - first_of_month = today.replace(day=1) - params = { - "accountNumber": self.number, - "startDate": first_of_month.strftime("%m/%d/%Y 12:00:00 AM"), - "endDate": today.strftime("%m/%d/%Y 11:59:59 PM"), - "OPCO": self.company.name, - "ServicePointNumber": await self.get_service_point_number(jwt), - "intervalBehavior": "Automatic", - } - async with session.get( - f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/" - f"MPUData/{self.number}/Daily", - headers=headers, - params=params, - ) as resp: - if resp.status != 200: - raise UsageDataFailure( - f"Failed to get month data: {resp.status} {headers}" - ) - else: + headers = {"Authorization": f"bearer {jwt}"} + today = datetime.datetime.now() + first_of_month = today.replace(day=1) + params = { + "accountNumber": self.number, + "startDate": first_of_month.strftime("%m/%d/%Y 12:00:00 AM"), + "endDate": today.strftime("%m/%d/%Y 11:59:59 PM"), + "OPCO": self.company.name, + "ServicePointNumber": self.service_point_number, + "intervalBehavior": "Automatic", + } + async with self.session.get( + f"https://customerservice2api.southerncompany.com/api/MyPowerUsage/" + f"MPUData/{self.number}/Daily", + headers=headers, + params=params, + ) as resp: + if resp.status != 200: + raise UsageDataFailure( + f"Failed to get month data: {resp.status} {headers}" + ) + else: + try: + connection = await resp.json() + except (ContentTypeError, json.JSONDecodeError) as err: try: - connection = await resp.json() - except (ContentTypeError, json.JSONDecodeError) as err: - try: - error_text = await resp.text() - except aiohttp.ClientError: - error_text = err.msg - raise CantReachSouthernCompany( - f"Incorrect mimetype while trying to get month data. {error_text}" - ) from err - data = connection["Data"] - return MonthlyUsage( - dollars_to_date=data["DollarsToDate"], - total_kwh_used=data["TotalkWhUsed"], - average_daily_usage=data["AverageDailyUsage"], - average_daily_cost=data["AverageDailyCost"], - projected_usage_low=data["ProjectedUsageLow"], - projected_usage_high=data["ProjectedUsageHigh"], - projected_bill_amount_low=data["ProjectedBillAmountLow"], - projected_bill_amount_high=data["ProjectedBillAmountHigh"], - ) + error_text = await resp.text() + except aiohttp.ClientError: + error_text = err.msg + raise CantReachSouthernCompany( + f"Incorrect mimetype while trying to get month data. {error_text}" + ) from err + data = connection["Data"] + return MonthlyUsage( + dollars_to_date=data["DollarsToDate"], + total_kwh_used=data["TotalkWhUsed"], + average_daily_usage=data["AverageDailyUsage"], + average_daily_cost=data["AverageDailyCost"], + projected_usage_low=data["ProjectedUsageLow"], + projected_usage_high=data["ProjectedUsageHigh"], + projected_bill_amount_low=data["ProjectedBillAmountLow"], + projected_bill_amount_high=data["ProjectedBillAmountHigh"], + ) diff --git a/src/southern_company_api/parser.py b/src/southern_company_api/parser.py index 6403e91..1e8cccc 100644 --- a/src/southern_company_api/parser.py +++ b/src/southern_company_api/parser.py @@ -26,12 +26,11 @@ async def get_request_verification_token(session: ClientSession) -> str: :return: the verification token """ try: - async with session: - http_response = await session.get( - "https://webauth.southernco.com/account/login" - ) - login_page = await http_response.text() - matches = re.findall(r'data-aft="(\S+)"', login_page) + http_response = await session.get( + "https://webauth.southernco.com/account/login" + ) + login_page = await http_response.text() + matches = re.findall(r'data-aft="(\S+)"', login_page) except Exception as error: raise CantReachSouthernCompany() from error if len(matches) < 1: @@ -113,18 +112,17 @@ async def _get_sc_web_token(self) -> str: "params": {"ReturnUrl": "null"}, } - async with self.session as session: - async with session.post( - "https://webauth.southernco.com/api/login", json=data, headers=headers - ) as response: - if response.status != 200: - raise CantReachSouthernCompany() - try: - connection = await response.json() - except (ContentTypeError, json.JSONDecodeError) as err: - raise InvalidLogin from err - if connection["statusCode"] == 500: - raise InvalidLogin() + async with self.session.post( + "https://webauth.southernco.com/api/login", json=data, headers=headers + ) as response: + if response.status != 200: + raise CantReachSouthernCompany() + try: + connection = await response.json() + except (ContentTypeError, json.JSONDecodeError) as err: + raise InvalidLogin from err + if connection["statusCode"] == 500: + raise InvalidLogin() sc_regex = re.compile(r"NAME='ScWebToken' value='(\S+.\S+.\S+)'", re.IGNORECASE) sc_data = sc_regex.search(connection["data"]["html"]) self._sc_expiry = datetime.datetime.now() + datetime.timedelta(hours=3) @@ -141,38 +139,37 @@ async def _get_southern_jwt_cookie(self) -> str: if await self.sc is None: raise CantReachSouthernCompany("Sc token cannot be refreshed") data = {"ScWebToken": self._sc} - async with self.session as session: - async with session.post( - "https://customerservice2.southerncompany.com/Account/LoginComplete?" - "ReturnUrl=null", - data=data, - ) as resp: - # Checking for unsuccessful login - if resp.status != 200: - raise NoScTokenFound( - f"Failed to get secondary ScWebToken: {resp.status} " - f"{resp.headers} {data}" - ) - # Regex to parse JWT out of headers - # NOTE: This used to be ScWebToken before 02/07/2023 - swtregex = re.compile(r"SouthernJwtCookie=(\S*);", re.IGNORECASE) - # Parsing response header to get token - swtcookies = resp.headers.get("set-cookie") - if swtcookies: - swtmatches = swtregex.search(swtcookies) - - # Checking for matches - if swtmatches and swtmatches.group(1): - swtoken = swtmatches.group(1) - else: - raise NoScTokenFound( - "Failed to get secondary ScWebToken: Could not find any " - "token matches in headers" - ) + async with self.session.post( + "https://customerservice2.southerncompany.com/Account/LoginComplete?" + "ReturnUrl=null", + data=data, + ) as resp: + # Checking for unsuccessful login + if resp.status != 200: + raise NoScTokenFound( + f"Failed to get secondary ScWebToken: {resp.status} " + f"{resp.headers} {data} sc_expiry: {self._sc_expiry}" + ) + # Regex to parse JWT out of headers + # NOTE: This used to be ScWebToken before 02/07/2023 + swtregex = re.compile(r"SouthernJwtCookie=(\S*);", re.IGNORECASE) + # Parsing response header to get token + swtcookies = resp.headers.get("set-cookie") + if swtcookies: + swtmatches = swtregex.search(swtcookies) + + # Checking for matches + if swtmatches and swtmatches.group(1): + swtoken = swtmatches.group(1) else: raise NoScTokenFound( - "Failed to get secondary ScWebToken: No cookies were sent back." + "Failed to get secondary ScWebToken: Could not find any " + "token matches in headers" ) + else: + raise NoScTokenFound( + "Failed to get secondary ScWebToken: No cookies were sent back." + ) return swtoken async def get_jwt(self) -> str: @@ -181,37 +178,34 @@ async def get_jwt(self) -> str: # Now fetch JWT after secondary ScWebToken # NOTE: This used to be ScWebToken before 02/07/2023 headers = {"Cookie": f"SouthernJwtCookie={swtoken}"} - async with self.session as session: - async with session.get( - "https://customerservice2.southerncompany.com/Account/LoginValidated/" - "JwtToken", - headers=headers, - ) as resp: - if resp.status != 200: - raise NoJwtTokenFound( - f"Failed to get JWT: {resp.status} {await resp.text()} " - f"{headers}" - ) - # Regex to parse JWT out of headers - regex = re.compile(r"ScJwtToken=(\S*);", re.IGNORECASE) - - # Parsing response header to get token - cookies = resp.headers.get("set-cookie") - if cookies: - matches = regex.search(cookies) - - # Checking for matches - if matches and matches.group(1): - token = matches.group(1) - else: - raise NoJwtTokenFound( - "Failed to get JWT: Could not find any token matches in " - "headers" - ) + async with self.session.get( + "https://customerservice2.southerncompany.com/Account/LoginValidated/" + "JwtToken", + headers=headers, + ) as resp: + if resp.status != 200: + raise NoJwtTokenFound( + f"Failed to get JWT: {resp.status} {await resp.text()} " + f"{headers}" + ) + # Regex to parse JWT out of headers + regex = re.compile(r"ScJwtToken=(\S*);", re.IGNORECASE) + + # Parsing response header to get token + cookies = resp.headers.get("set-cookie") + if cookies: + matches = regex.search(cookies) + + # Checking for matches + if matches and matches.group(1): + token = matches.group(1) else: raise NoJwtTokenFound( - "Failed to get JWT: No cookies were sent back." + "Failed to get JWT: Could not find any token matches in " + "headers" ) + else: + raise NoJwtTokenFound("Failed to get JWT: No cookies were sent back.") # Returning JWT self._jwt = token @@ -221,37 +215,39 @@ async def get_jwt(self) -> str: return token async def get_accounts(self) -> List[Account]: + print("AHAHA") if await self.jwt is None: raise CantReachSouthernCompany("Can't get jwt. Expired and not refreshed") headers = {"Authorization": f"bearer {self._jwt}"} - async with self.session as session: - async with session.get( - "https://customerservice2api.southerncompany.com/api/account/" - "getAllAccounts", - headers=headers, - ) as resp: - if resp.status != 200: - raise AccountFailure("failed to get accounts") + async with self.session.get( + "https://customerservice2api.southerncompany.com/api/account/" + "getAllAccounts", + headers=headers, + ) as resp: + if resp.status != 200: + raise AccountFailure("failed to get accounts") + try: + account_json = await resp.json() + except (ContentTypeError, json.JSONDecodeError) as err: try: - account_json = await resp.json() - except (ContentTypeError, json.JSONDecodeError) as err: - try: - error_text = await resp.text() - except aiohttp.ClientError: - error_text = err.msg - raise CantReachSouthernCompany( - f"Incorrect mimetype while trying to get accounts. {error_text}" - ) from err - accounts = [] - for account in account_json["Data"]: - accounts.append( - Account( - name=account["Description"], - primary=account["PrimaryAccount"] == "Y", - number=account["AccountNumber"], - company=COMPANY_MAP.get(account["Company"], Company.GPC), - session=self.session, - ) + error_text = await resp.text() + except aiohttp.ClientError: + error_text = err.msg + raise CantReachSouthernCompany( + f"Incorrect mimetype while trying to get accounts. {error_text}" + ) from err + accounts = [] + for account in account_json["Data"]: + accounts.append( + Account( + name=account["Description"], + primary=account["PrimaryAccount"] == "Y", + number=account["AccountNumber"], + company=COMPANY_MAP.get(account["Company"], Company.GPC), + session=self.session, ) + ) + for account in accounts: + await account.get_service_point_number(self._jwt) self._accounts = accounts return accounts diff --git a/tests/test_account.py b/tests/test_account.py index be56caa..9acd70c 100644 --- a/tests/test_account.py +++ b/tests/test_account.py @@ -8,44 +8,48 @@ from tests import MockResponse, test_get_hourly_usage, test_get_month_data -def test_can_create(): - Account("sample", True, "1", Company.GPC, aiohttp.ClientSession()) +@pytest.mark.asyncio +async def test_can_create(): + async with aiohttp.ClientSession() as session: + Account("sample", True, "1", Company.GPC, session) @pytest.mark.asyncio async def test_get_hourly_data(): - acc = Account("sample", True, "1", Company.GPC, aiohttp.ClientSession()) - with patch( - "src.southern_company_api.account.aiohttp.ClientSession.get" - ) as mock_get, patch( - "southern_company_api.account.Account.get_service_point_number" - ) as mock_get_service_point: - mock_get.return_value = MockResponse("", 200, "", test_get_hourly_usage) - mock_get_service_point.return_value.__aenter__.return_value = "" - await acc.get_hourly_data( - datetime.datetime.now() - datetime.timedelta(days=3), - datetime.datetime.now() - datetime.timedelta(days=2, hours=22), - "dummy_jwt", - ) - assert len(list(acc.hourly_data.values())) == 2 + async with aiohttp.ClientSession() as session: + acc = Account("sample", True, "1", Company.GPC, session) + with patch( + "src.southern_company_api.account.aiohttp.ClientSession.get" + ) as mock_get, patch( + "southern_company_api.account.Account.get_service_point_number" + ) as mock_get_service_point: + mock_get.return_value = MockResponse("", 200, "", test_get_hourly_usage) + mock_get_service_point.return_value.__aenter__.return_value = "" + await acc.get_hourly_data( + datetime.datetime.now() - datetime.timedelta(days=3), + datetime.datetime.now() - datetime.timedelta(days=2, hours=22), + "dummy_jwt", + ) + assert len(list(acc.hourly_data.values())) == 2 @pytest.mark.asyncio async def test_ga_power_get_monthly_data(): - acc = Account("sample", True, "1", Company.GPC, aiohttp.ClientSession()) - with patch( - "src.southern_company_api.account.aiohttp.ClientSession.get" - ) as mock_get, patch( - "southern_company_api.account.Account.get_service_point_number" - ) as mock_get_service_point: - mock_get.return_value = MockResponse("", 200, "", test_get_month_data) - mock_get_service_point.return_value.__aenter__.return_value = "" - month = await acc.get_month_data("dummy_jwt") - assert month.total_kwh_used == 97.0 - assert month.dollars_to_date == 13.974766406622413 - assert month.average_daily_cost == 2.79 - assert month.average_daily_usage == 19.17 - assert month.projected_usage_high == 629.0 - assert month.projected_usage_low == 419.0 - assert month.projected_bill_amount_high == 91.0 - assert month.projected_bill_amount_low == 60.0 + async with aiohttp.ClientSession() as session: + acc = Account("sample", True, "1", Company.GPC, session) + with patch( + "src.southern_company_api.account.aiohttp.ClientSession.get" + ) as mock_get, patch( + "southern_company_api.account.Account.get_service_point_number" + ) as mock_get_service_point: + mock_get.return_value = MockResponse("", 200, "", test_get_month_data) + mock_get_service_point.return_value.__aenter__.return_value = "" + month = await acc.get_month_data("dummy_jwt") + assert month.total_kwh_used == 97.0 + assert month.dollars_to_date == 13.974766406622413 + assert month.average_daily_cost == 2.79 + assert month.average_daily_usage == 19.17 + assert month.projected_usage_high == 629.0 + assert month.projected_usage_low == 419.0 + assert month.projected_bill_amount_high == 91.0 + assert month.projected_bill_amount_low == 60.0 diff --git a/tests/test_parser.py b/tests/test_parser.py index 21d84f1..7c48a5d 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -26,14 +26,17 @@ ) -def test_can_create(): - SouthernCompanyAPI("user", "pass", aiohttp.ClientSession()) +@pytest.mark.asyncio +async def test_can_create(): + async with aiohttp.ClientSession() as session: + SouthernCompanyAPI("user", "pass", session) @pytest.mark.asyncio async def test_get_request_verification_token(): - token = await get_request_verification_token(aiohttp.ClientSession()) - assert len(token) > 1 + async with aiohttp.ClientSession() as session: + token = await get_request_verification_token(session) + assert len(token) > 1 # @pytest.mark.asyncio @@ -55,7 +58,8 @@ async def test_cant_find_request_token(): "src.southern_company_api.parser.aiohttp.ClientResponse.text", return_value="" ): with pytest.raises(NoRequestTokenFound): - await get_request_verification_token(aiohttp.ClientSession()) + async with aiohttp.ClientSession() as session: + await get_request_verification_token(session) @pytest.mark.asyncio @@ -67,32 +71,35 @@ async def test_can_authenticate(): ) as mock__get_sc_web_token: mock_get_request_verification_token.return_value = "fake_token" mock__get_sc_web_token.return_value = "fake_sc" - api = SouthernCompanyAPI("", "", aiohttp.ClientSession()) - result = await api.authenticate() - assert result is True - mock_get_request_verification_token.assert_called_once() - mock__get_sc_web_token.assert_called_once() + async with aiohttp.ClientSession() as session: + api = SouthernCompanyAPI("", "", session) + result = await api.authenticate() + assert result is True + mock_get_request_verification_token.assert_called_once() + mock__get_sc_web_token.assert_called_once() @pytest.mark.asyncio async def test_ga_power_get_sc_web_token(): with patch("southern_company_api.parser.aiohttp.ClientSession.post") as mock_post: mock_post.return_value = MockResponse("", 200, "", ga_power_sample_sc_response) - sca = SouthernCompanyAPI("", "", aiohttp.ClientSession()) - sca._request_token = "sample" - response_token = await sca._get_sc_web_token() - assert response_token == "sample_sc_token" + async with aiohttp.ClientSession() as session: + sca = SouthernCompanyAPI("", "", session) + sca._request_token = "sample" + response_token = await sca._get_sc_web_token() + assert response_token == "sample_sc_token" @pytest.mark.asyncio async def test_get_sc_web_token_wrong_login(): - sca = SouthernCompanyAPI("user", "pass", aiohttp.ClientSession()) - with patch( - "src.southern_company_api.parser.aiohttp.ClientSession.post" - ) as mock_post: - mock_post.return_value = MockResponse("", 200, "", {"statusCode": 500}) - with pytest.raises(InvalidLogin): - await sca._get_sc_web_token() + async with aiohttp.ClientSession() as session: + sca = SouthernCompanyAPI("user", "pass", session) + with patch( + "src.southern_company_api.parser.aiohttp.ClientSession.post" + ) as mock_post: + mock_post.return_value = MockResponse("", 200, "", {"statusCode": 500}) + with pytest.raises(InvalidLogin): + await sca._get_sc_web_token() @pytest.mark.asyncio @@ -103,11 +110,12 @@ async def test_ga_power_get_jwt_cookie(): mock_post.return_value = MockResponse( "", 200, ga_power_southern_jwt_cookie_header, "" ) - sca = SouthernCompanyAPI("", "", aiohttp.ClientSession()) - sca._sc = "" - sca._sc_expiry = datetime.datetime.now() + datetime.timedelta(hours=3) - token = await sca._get_southern_jwt_cookie() - assert token == "sample_cookie" + async with aiohttp.ClientSession() as session: + sca = SouthernCompanyAPI("", "", session) + sca._sc = "" + sca._sc_expiry = datetime.datetime.now() + datetime.timedelta(hours=3) + token = await sca._get_southern_jwt_cookie() + assert token == "sample_cookie" @pytest.mark.asyncio @@ -119,14 +127,19 @@ async def test_ga_power_get_jwt(): ) as mock_get_cookie: mock_get.return_value = MockResponse("", 200, ga_power_jwt_header, "") mock_get_cookie.return_value.__aenter__.return_value = "" - sca = SouthernCompanyAPI("", "", aiohttp.ClientSession()) - token = await sca.get_jwt() - assert token == "sample_jwt" + async with aiohttp.ClientSession() as session: + sca = SouthernCompanyAPI("", "", session) + token = await sca.get_jwt() + assert token == "sample_jwt" @pytest.mark.asyncio async def test_ga_power_get_accounts(): - with patch("src.southern_company_api.parser.aiohttp.ClientSession.get") as mock_get: + with patch( + "src.southern_company_api.parser.aiohttp.ClientSession.get" + ) as mock_get, patch( + "src.southern_company_api.parser.Account.get_service_point_number" + ): mock_get.return_value.__aenter__.return_value.json.return_value = ( ga_power_sample_account_response ) @@ -137,7 +150,8 @@ async def mock_jwt(_foo_self: SouthernCompanyAPI) -> str: return "" with patch.object(SouthernCompanyAPI, "jwt", new=mock_jwt): - sca = SouthernCompanyAPI("", "", aiohttp.ClientSession()) - response_token: typing.List[Account] = await sca.get_accounts() - assert response_token[0].name == "Home Energy" - assert sca._accounts == response_token + async with aiohttp.ClientSession() as session: + sca = SouthernCompanyAPI("", "", session) + response_token: typing.List[Account] = await sca.get_accounts() + assert response_token[0].name == "Home Energy" + assert sca._accounts == response_token