test(login): satisfy type and coverage gates
This commit is contained in:
@@ -23,6 +23,7 @@ import time
|
|||||||
import urllib.error
|
import urllib.error
|
||||||
import urllib.request
|
import urllib.request
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from ..paths import bot_bottle_root
|
from ..paths import bot_bottle_root
|
||||||
|
|
||||||
@@ -49,7 +50,7 @@ def _flag(argv: list[str], name: str) -> str | None:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def _post(url: str, payload: dict) -> dict:
|
def _post(url: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||||||
data = json.dumps(payload).encode()
|
data = json.dumps(payload).encode()
|
||||||
req = urllib.request.Request(
|
req = urllib.request.Request(
|
||||||
url, data=data, headers={"Content-Type": "application/json"}
|
url, data=data, headers={"Content-Type": "application/json"}
|
||||||
@@ -58,7 +59,7 @@ def _post(url: str, payload: dict) -> dict:
|
|||||||
return json.loads(resp.read())
|
return json.loads(resp.read())
|
||||||
|
|
||||||
|
|
||||||
def _get(url: str) -> tuple[int, dict]:
|
def _get(url: str) -> tuple[int, dict[str, Any]]:
|
||||||
req = urllib.request.Request(url)
|
req = urllib.request.Request(url)
|
||||||
try:
|
try:
|
||||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||||
|
|||||||
@@ -6,6 +6,9 @@ import json
|
|||||||
import os
|
import os
|
||||||
import tempfile
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
|
import urllib.error
|
||||||
|
from email.message import Message
|
||||||
|
from typing import Any
|
||||||
from unittest.mock import MagicMock, patch
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
|
||||||
@@ -16,7 +19,9 @@ class TestFlagParsing(unittest.TestCase):
|
|||||||
|
|
||||||
def test_console_url_equals_form(self) -> None:
|
def test_console_url_equals_form(self) -> None:
|
||||||
from bot_bottle.cli.login import _flag
|
from bot_bottle.cli.login import _flag
|
||||||
self.assertEqual(_flag(["--console-url=http://x"], "--console-url"), "http://x")
|
self.assertEqual(
|
||||||
|
_flag(["--console-url=http://x"], "--console-url"), "http://x"
|
||||||
|
)
|
||||||
|
|
||||||
def test_label_flag(self) -> None:
|
def test_label_flag(self) -> None:
|
||||||
from bot_bottle.cli.login import _flag
|
from bot_bottle.cli.login import _flag
|
||||||
@@ -27,6 +32,42 @@ class TestFlagParsing(unittest.TestCase):
|
|||||||
self.assertIsNone(_flag([], "--console-url"))
|
self.assertIsNone(_flag([], "--console-url"))
|
||||||
|
|
||||||
|
|
||||||
|
class TestHttpHelpers(unittest.TestCase):
|
||||||
|
def test_post_sends_json_and_decodes_response(self) -> None:
|
||||||
|
from bot_bottle.cli.login import _post
|
||||||
|
|
||||||
|
response = MagicMock()
|
||||||
|
response.__enter__.return_value.read.return_value = b'{"ok": true}'
|
||||||
|
with patch("urllib.request.urlopen", return_value=response) as urlopen:
|
||||||
|
self.assertEqual(
|
||||||
|
_post("http://console/start", {"label": "host"}), {"ok": True}
|
||||||
|
)
|
||||||
|
|
||||||
|
request = urlopen.call_args.args[0]
|
||||||
|
self.assertEqual(request.data, b'{"label": "host"}')
|
||||||
|
self.assertEqual(request.get_header("Content-type"), "application/json")
|
||||||
|
|
||||||
|
def test_get_decodes_success_response(self) -> None:
|
||||||
|
from bot_bottle.cli.login import _get
|
||||||
|
|
||||||
|
response = MagicMock()
|
||||||
|
response.__enter__.return_value.status = 200
|
||||||
|
response.__enter__.return_value.read.return_value = b'{"status": "pending"}'
|
||||||
|
with patch("urllib.request.urlopen", return_value=response):
|
||||||
|
self.assertEqual(
|
||||||
|
_get("http://console/status"), (200, {"status": "pending"})
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_get_returns_http_error_status(self) -> None:
|
||||||
|
from bot_bottle.cli.login import _get
|
||||||
|
|
||||||
|
error = urllib.error.HTTPError(
|
||||||
|
"http://console/status", 410, "gone", Message(), None
|
||||||
|
)
|
||||||
|
with patch("urllib.request.urlopen", side_effect=error):
|
||||||
|
self.assertEqual(_get("http://console/status"), (410, {}))
|
||||||
|
|
||||||
|
|
||||||
class TestSaveCredentials(unittest.TestCase):
|
class TestSaveCredentials(unittest.TestCase):
|
||||||
def test_writes_json_and_sets_perms(self) -> None:
|
def test_writes_json_and_sets_perms(self) -> None:
|
||||||
from bot_bottle.cli.login import _save_credentials
|
from bot_bottle.cli.login import _save_credentials
|
||||||
@@ -50,7 +91,9 @@ class TestSaveCredentials(unittest.TestCase):
|
|||||||
tmp_perms_at_replace: list[int] = []
|
tmp_perms_at_replace: list[int] = []
|
||||||
real_replace = os.replace
|
real_replace = os.replace
|
||||||
|
|
||||||
def _spy_replace(src: "str | os.PathLike", dst: "str | os.PathLike") -> None:
|
def _spy_replace(
|
||||||
|
src: str | os.PathLike[str], dst: str | os.PathLike[str]
|
||||||
|
) -> None:
|
||||||
tmp_perms_at_replace.append(_Path(src).stat().st_mode & 0o777)
|
tmp_perms_at_replace.append(_Path(src).stat().st_mode & 0o777)
|
||||||
real_replace(src, dst)
|
real_replace(src, dst)
|
||||||
|
|
||||||
@@ -64,6 +107,11 @@ class TestSaveCredentials(unittest.TestCase):
|
|||||||
|
|
||||||
|
|
||||||
class TestCmdLoginMissingUrl(unittest.TestCase):
|
class TestCmdLoginMissingUrl(unittest.TestCase):
|
||||||
|
def test_help_returns_0(self) -> None:
|
||||||
|
from bot_bottle.cli.login import cmd_login
|
||||||
|
|
||||||
|
self.assertEqual(cmd_login(["--help"]), 0)
|
||||||
|
|
||||||
def test_returns_1_without_url(self) -> None:
|
def test_returns_1_without_url(self) -> None:
|
||||||
from bot_bottle.cli.login import cmd_login
|
from bot_bottle.cli.login import cmd_login
|
||||||
with patch.dict(os.environ, {}, clear=True):
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
@@ -75,18 +123,26 @@ class TestCmdLoginMissingUrl(unittest.TestCase):
|
|||||||
"""Exits 1 (network error) not because of missing URL when env var is set."""
|
"""Exits 1 (network error) not because of missing URL when env var is set."""
|
||||||
from bot_bottle.cli.login import cmd_login
|
from bot_bottle.cli.login import cmd_login
|
||||||
|
|
||||||
def _fail_post(url, payload):
|
def _fail_post(_url: str, _payload: dict[str, Any]) -> dict[str, Any]:
|
||||||
raise OSError("connection refused")
|
raise OSError("connection refused")
|
||||||
|
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
with patch.dict(os.environ, {"BB_CONSOLE_URL": "http://localhost:9999", "BOT_BOTTLE_ROOT": tmp}):
|
with patch.dict(
|
||||||
|
os.environ,
|
||||||
|
{
|
||||||
|
"BB_CONSOLE_URL": "http://localhost:9999",
|
||||||
|
"BOT_BOTTLE_ROOT": tmp,
|
||||||
|
},
|
||||||
|
):
|
||||||
with patch("bot_bottle.cli.login._post", side_effect=_fail_post):
|
with patch("bot_bottle.cli.login._post", side_effect=_fail_post):
|
||||||
result = cmd_login([])
|
result = cmd_login([])
|
||||||
self.assertEqual(result, 1)
|
self.assertEqual(result, 1)
|
||||||
|
|
||||||
|
|
||||||
class TestCmdLoginFlow(unittest.TestCase):
|
class TestCmdLoginFlow(unittest.TestCase):
|
||||||
def _run_with_mocks(self, poll_responses: list[dict], tmp: str) -> int:
|
def _run_with_mocks(
|
||||||
|
self, poll_responses: list[dict[str, Any]], tmp: str
|
||||||
|
) -> int:
|
||||||
from bot_bottle.cli.login import cmd_login
|
from bot_bottle.cli.login import cmd_login
|
||||||
|
|
||||||
start_resp = {
|
start_resp = {
|
||||||
@@ -98,10 +154,12 @@ class TestCmdLoginFlow(unittest.TestCase):
|
|||||||
|
|
||||||
poll_iter = iter(poll_responses)
|
poll_iter = iter(poll_responses)
|
||||||
|
|
||||||
def _fake_post(url, payload):
|
def _fake_post(
|
||||||
|
_url: str, _payload: dict[str, Any]
|
||||||
|
) -> dict[str, Any]:
|
||||||
return start_resp
|
return start_resp
|
||||||
|
|
||||||
def _fake_get(url):
|
def _fake_get(_url: str) -> tuple[int, dict[str, Any]]:
|
||||||
try:
|
try:
|
||||||
resp = next(poll_iter)
|
resp = next(poll_iter)
|
||||||
except StopIteration:
|
except StopIteration:
|
||||||
@@ -170,10 +228,15 @@ class TestCmdLoginFlow(unittest.TestCase):
|
|||||||
}
|
}
|
||||||
poll_iter = iter([{"status": "pending"}, approved])
|
poll_iter = iter([{"status": "pending"}, approved])
|
||||||
|
|
||||||
|
def _fake_get(_url: str) -> tuple[int, dict[str, str]]:
|
||||||
|
return 200, next(poll_iter)
|
||||||
|
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
|
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
|
||||||
with patch("bot_bottle.cli.login._post", return_value=start_resp):
|
with patch("bot_bottle.cli.login._post", return_value=start_resp):
|
||||||
with patch("bot_bottle.cli.login._get", side_effect=lambda u: (200, next(poll_iter))):
|
with patch(
|
||||||
|
"bot_bottle.cli.login._get", side_effect=_fake_get
|
||||||
|
):
|
||||||
with patch("time.sleep") as mock_sleep:
|
with patch("time.sleep") as mock_sleep:
|
||||||
result = cmd_login(["--console-url", "http://console"])
|
result = cmd_login(["--console-url", "http://console"])
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user