Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion requirements_test.txt
Original file line number Diff line number Diff line change
Expand Up @@ -4,5 +4,4 @@ pytest-asyncio
pytest-cov
pytest-timeout
aiohttp
aioresponses
freezegun
139 changes: 85 additions & 54 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,99 @@
"""Provide common pytest fixtures."""
import re
from unittest.mock import patch

import pytest
from aioresponses import aioresponses
from yarl import URL

import openeihttp

pytestmark = pytest.mark.asyncio


class MockResponse:
"""Mock aiohttp response."""

def __init__(self, status: int, text: str) -> None:
"""Initialize."""
self.status = status
self._text = text

async def text(self) -> str:
"""Return text."""
return self._text

async def read(self) -> bytes:
"""Return bytes."""
return self._text.encode("utf-8")

async def __aenter__(self):
"""Enter context."""
return self

async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
"""Exit context."""
pass


class AiohttpClientMocker:
"""Mock aiohttp client requests."""

def __init__(self) -> None:
"""Initialize."""
self.mocks = []
self._patcher = patch("aiohttp.ClientSession._request", side_effect=self._mock_request)

def get(self, url, status=200, body="", repeat=False, exception=None, **kwargs) -> None:
"""Register a mock GET request."""
self.mocks.append(
{
"method": "GET",
"url": url,
"status": status,
"body": body,
"repeat": repeat,
"exception": exception,
}
)

def start(self) -> None:
"""Start patching."""
self._patcher.start()

def stop(self) -> None:
"""Stop patching."""
self._patcher.stop()

async def _mock_request(self, method, url, *args, **kwargs):
"""Intercept and mock requests."""
params = kwargs.get("params")
url_str = str(URL(url).with_query(params)) if params else str(url)

for mock in self.mocks:
if mock["method"] == method:
pattern = mock["url"]
matched = False
if isinstance(pattern, re.Pattern):
if pattern.search(url_str):
matched = True
elif pattern == url_str:
matched = True

if matched:
if not mock["repeat"]:
self.mocks.remove(mock)
if mock["exception"] is not None:
raise mock["exception"]
return MockResponse(mock["status"], mock["body"])
raise AssertionError(f"No mock registered for {method} {url_str}")


@pytest.fixture
def mock_aioclient():
"""Fixture to mock aioclient calls."""
with aioresponses() as m:
yield m
mocker = AiohttpClientMocker()
mocker.start()
yield mocker
mocker.stop()


@pytest.fixture(name="test_lookup")
Expand Down Expand Up @@ -49,56 +130,6 @@ def test_lookup_tier_low():
)


@pytest.fixture(name="test_lookup_tier_med")
def test_lookup_tier_med():
"""Load the charger data."""
return openeihttp.Rates(
api="fakeAPIKey",
reading="10.3",
plan="574613aa5457a3557e906f5b",
)


@pytest.fixture(name="test_lookup_tier_high")
def test_lookup_tier_high():
"""Load the charger data."""
return openeihttp.Rates(
api="fakeAPIKey",
reading="40.1",
plan="574613aa5457a3557e906f5b",
)


@pytest.fixture(name="test_lookup_monthly_tier_low")
def test_lookup_monthly_tier_low():
"""Load the charger data."""
return openeihttp.Rates(
api="fakeAPIKey",
reading="114",
plan="574613aa5457a3557e906f5b",
)


@pytest.fixture(name="test_lookup_monthly_tier_med")
def test_lookup_monthly_tier_med():
"""Load the charger data."""
return openeihttp.Rates(
api="fakeAPIKey",
reading="301",
plan="574613aa5457a3557e906f5b",
)


@pytest.fixture(name="test_lookup_monthly_tier_high")
def test_lookup_monthly_tier_high():
"""Load the charger data."""
return openeihttp.Rates(
api="fakeAPIKey",
reading="1300",
plan="574613aa5457a3557e906f5b",
)


@pytest.fixture(name="test_rates")
def test_rates():
"""Load the charger data."""
Expand Down
Loading