GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/hve-core

Public

mirrored from https://github.com/microsoft/hve-coreAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
docs/transparency-note

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

hve-core/.github/skills/experimental/mural/tests

.github/skills/experimental/mural/tests/conftest.py

218lines · modecode

RawDownload
Latest commit unavailable.
1# Copyright (c) Microsoft Corporation.
2# SPDX-License-Identifier: MIT
3"""Shared fixtures for Mural skill tests."""
4
5from __future__ import annotations
6
7import io
8import os
9import pathlib
10import sys
11import urllib.error
12from collections.abc import Callable
13from dataclasses import dataclass, field
14from email.message import Message
15from typing import Any, Literal
16
17import pytest
18from test_constants import (
19 ENV_BASE_URL,
20 ENV_CLIENT_ID,
21 ENV_CLIENT_SECRET,
22 ENV_REDIRECT_URI,
23 ENV_TOKEN_STORE,
24 MURAL_ENV_VARS,
25 TEST_BASE_URL,
26 TEST_CLIENT_ID,
27 TEST_CLIENT_SECRET,
28 TEST_REDIRECT_URI,
29)
30
31
32class FakeHttpResponse:
33 """Minimal HTTP response stub mirroring `urllib` context-manager semantics."""
34
35 def __init__(
36 self,
37 body: bytes | str = b"",
38 *,
39 status: int = 200,
40 headers: dict[str, str] | None = None,
41 ) -> None:
42 if isinstance(body, str):
43 body = body.encode("utf-8")
44 self._body = body
45 self.status = status
46 msg = Message()
47 for key, value in (headers or {}).items():
48 msg[key] = value
49 self.headers = msg
50
51 def __enter__(self) -> "FakeHttpResponse":
52 return self
53
54 def __exit__(self, exc_type: object, exc: object, tb: object) -> Literal[False]:
55 return False
56
57 def read(self, amt: int | None = None) -> bytes:
58 if amt is None or amt < 0:
59 return self._body
60 return self._body[:amt]
61
62
63@dataclass
64class RecordedHttpCall:
65 """Captured HTTP request issued through the `_http` seam."""
66
67 method: str
68 url: str
69 headers: dict[str, str]
70 data: bytes | None
71
72
73@dataclass
74class HttpRecorder:
75 """Callable seam that records requests and returns queued responses."""
76
77 responses: list[Any] = field(default_factory=list)
78 calls: list[RecordedHttpCall] = field(default_factory=list)
79
80 def __call__(self, request: Any, *args: Any, **kwargs: Any) -> Any:
81 method = getattr(request, "get_method", lambda: "GET")()
82 url = getattr(request, "full_url", getattr(request, "url", ""))
83 raw_headers = getattr(request, "headers", {}) or {}
84 headers = {str(k): str(v) for k, v in dict(raw_headers).items()}
85 data = getattr(request, "data", None)
86 self.calls.append(
87 RecordedHttpCall(method=method, url=url, headers=headers, data=data)
88 )
89 if not self.responses:
90 raise AssertionError(f"no queued response for {method} {url}")
91 outcome = self.responses.pop(0)
92 if isinstance(outcome, BaseException):
93 raise outcome
94 return outcome
95
96
97ResponseFactory = Callable[..., FakeHttpResponse]
98HttpErrorFactory = Callable[..., urllib.error.HTTPError]
99StdinFactory = Callable[[str], None]
100
101
102@pytest.fixture(autouse=True)
103def reset_environment(monkeypatch: pytest.MonkeyPatch) -> None:
104 """Strip Mural env vars and seed deterministic defaults."""
105 for var in MURAL_ENV_VARS:
106 monkeypatch.delenv(var, raising=False)
107 monkeypatch.setenv(ENV_BASE_URL, TEST_BASE_URL)
108 monkeypatch.setenv(ENV_CLIENT_ID, TEST_CLIENT_ID)
109 monkeypatch.setenv(ENV_CLIENT_SECRET, TEST_CLIENT_SECRET)
110 monkeypatch.setenv(ENV_REDIRECT_URI, TEST_REDIRECT_URI)
111
112
113@pytest.fixture
114def mural_module() -> Any:
115 """Import the `mural` package fresh for each test.
116
117 Purges every previously-imported ``mural`` and ``mural.*`` submodule from
118 ``sys.modules`` before re-importing so that module-level state (env vars,
119 cached config) is re-evaluated under the active monkeypatch.
120 """
121 targets = {
122 name
123 for name in list(sys.modules)
124 if name == "mural" or name.startswith("mural.")
125 }
126 for name in targets:
127 sys.modules.pop(name, None)
128
129 import mural
130
131 return mural
132
133
134@pytest.fixture
135def fake_token_store(
136 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
137) -> pathlib.Path:
138 """Provide a temp token-store path with mode 0600 and wire `MURAL_TOKEN_STORE`."""
139 store_path = tmp_path / "mural-token.json"
140 fd = os.open(str(store_path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
141 try:
142 os.write(fd, b"{}")
143 finally:
144 os.close(fd)
145 os.chmod(store_path, 0o600)
146 monkeypatch.setenv(ENV_TOKEN_STORE, str(store_path))
147 return store_path
148
149
150@pytest.fixture
151def fake_now() -> Callable[[], float]:
152 """Return a deterministic `time.time()` replacement."""
153 state = {"value": 1_700_000_000.0}
154
155 def _now() -> float:
156 return state["value"]
157
158 _now.advance = lambda delta: state.update(value=state["value"] + float(delta)) # type: ignore[attr-defined]
159 _now.set = lambda value: state.update(value=float(value)) # type: ignore[attr-defined]
160 return _now
161
162
163@pytest.fixture
164def response_factory() -> ResponseFactory:
165 """Return a factory for `FakeHttpResponse` objects."""
166
167 def _factory(
168 body: bytes | str = b"",
169 *,
170 status: int = 200,
171 headers: dict[str, str] | None = None,
172 ) -> FakeHttpResponse:
173 return FakeHttpResponse(body, status=status, headers=headers)
174
175 return _factory
176
177
178@pytest.fixture
179def http_error_factory() -> HttpErrorFactory:
180 """Return a factory for `urllib.error.HTTPError` instances."""
181
182 def _factory(
183 body: bytes | str = b"",
184 *,
185 code: int = 400,
186 url: str = f"{TEST_BASE_URL}/test",
187 headers: dict[str, str] | None = None,
188 ) -> urllib.error.HTTPError:
189 if isinstance(body, str):
190 body = body.encode("utf-8")
191 msg = Message()
192 for key, value in (headers or {}).items():
193 msg[key] = value
194 return urllib.error.HTTPError(
195 url=url,
196 code=code,
197 msg="error",
198 hdrs=msg,
199 fp=io.BytesIO(body),
200 )
201
202 return _factory
203
204
205@pytest.fixture
206def recorded_http() -> HttpRecorder:
207 """Return a recording HTTP seam that can be injected via `_http=...`."""
208 return HttpRecorder()
209
210
211@pytest.fixture
212def stdin_factory(monkeypatch: pytest.MonkeyPatch) -> StdinFactory:
213 """Return a helper that replaces `sys.stdin` with text content."""
214
215 def _factory(text: str) -> None:
216 monkeypatch.setattr("sys.stdin", io.StringIO(text))
217
218 return _factory
219