added timestamp fields on all events

src/_testing/http_server.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+import json
+import threading
+from dataclasses import dataclass, field
+from typing import Any
+
+import flask
+from werkzeug.serving import make_server
+
+
+class SpyRequest:
+    def __init__(self, request: flask.Request) -> None:
+        self._method = request.method
+        self._path = request.path
+        self._query = request.query_string.decode()
+        self._bytes = request.get_data()
+
+    def method(self) -> str:
+        return self._method
+
+    def path(self) -> str:
+        return self._path
+
+    def query_string(self) -> str:
+        return self._query
+
+    def json(self) -> dict[str, Any]:
+        return json.loads(self._bytes.decode())
+
+    def text(self) -> str:
+        return self._bytes.decode()
+
+
+@dataclass
+class Spy:
+    received: list[SpyRequest] = field(default_factory=list)
+
+    def count_received(self) -> int:
+        return len(self.received)
+
+    def expect_request(self) -> SpyRequest:
+        if self.count_received() > 1:
+            raise AssertionError("Expected only one request")
+        if self.count_received() == 0:
+            raise AssertionError("Expected one request")
+        return self.received[0]
+
+
+class EmbeddedTestServer:
+    def __init__(
+        self,
+        spy: Spy,
+        path: str = "/webhooks/TestWebhook",
+        host: str = "127.0.0.1",
+        port: int = 8000,
+    ) -> None:
+        self.spy = spy
+        self.thread = self.ServerThread(self.create_app(spy, path), host, port)
+
+    def start(self) -> None:
+        self.thread.start()
+
+    def stop(self) -> None:
+        self.thread.shutdown()
+
+    def __enter__(self) -> "EmbeddedTestServer":
+        self.start()
+        return self
+
+    def __exit__(self, *args: object, **kwargs: object) -> None:
+        self.stop()
+
+    def create_app(self, spy: Spy, path: str):
+        app = flask.Flask("test-app")
+
+        @app.route(path, methods=["POST"])
+        def _() -> dict[str, str]:
+            spy.received.append(SpyRequest(request=flask.request))
+            return {"status": "OK"}
+
+        return app
+
+    class ServerThread(threading.Thread):
+        def __init__(self, app: flask.Flask, host: str, port: int):
+            threading.Thread.__init__(self)
+            self.server = make_server(host, port, app)
+            self.ctx = app.app_context()
+            self.ctx.push()
+
+        def run(self):
+            self.server.serve_forever()
+
+        def shutdown(self):
+            self.server.shutdown()

tests/collect/_utils.py → src/_testing/setup.py

@@ -35,37 +35,53 @@ def make_testfile(self, filename: str, content: str) -> Path:
         else:
             raise ValueError("Filename must end with '.py'")
         kwargs = {filename: content}
-        return self.test_dir.makepyfile(**kwargs)
+        return self.test_dir.makepyfile(**kwargs)  # pyright: ignore[reportUnknownMemberType]
 
-    def filter_traceback(
-        self, data: dict[str, Any] | list[Any]
-    ) -> dict[str, Any] | list[Any]:
-        return self._filter_traceback(deepcopy(data))
+    def sanitize(self, data: dict[str, Any] | list[Any]) -> dict[str, Any] | list[Any]:
+        return self._sanitize(deepcopy(data))
 
-    def _filter_traceback(
+    def _sanitize(
         self,
         data: Any,
     ) -> Any:
         if isinstance(data, dict):
-            for key, value in data.items():
+            for key, value in data.items():  # pyright: ignore[reportUnknownVariableType]
                 if key == "traceback":
                     data[key] = {
                         **data[key],
                         "entries": [
                             line
-                            for line in value["entries"]
+                            for line in value["entries"]  # pyright: ignore[reportUnknownVariableType]
                             if (
                                 "importlib" not in line["path"]
                                 and "pytest_asyncio" not in line["path"]
                             )
                         ],
                     }
+                elif key == "duration":
+                    data[key] = "omitted"
+                elif key == "total_duration":
+                    data[key] = "omitted"
+                elif key == "start_timestamp":
+                    data[key] = "omitted"
+                elif key == "stop_timestamp":
+                    data[key] = "omitted"
+                elif key == "timestamp":
+                    data[key] = "omitted"
+                elif key == "platform":
+                    data[key] = "omitted"
+                elif key == "processor":
+                    data[key] = "omitted"
+                elif key == "session_id":
+                    data[key] = "omitted"
+                elif key == "packages":
+                    data[key] = {}
                 elif isinstance(value, dict):
-                    data[key] = self._filter_traceback(value)
+                    data[key] = self._sanitize(value)
                 elif isinstance(value, list):
-                    data[key] = self._filter_traceback(value)
+                    data[key] = self._sanitize(value)
         elif isinstance(data, list):
-            data = [self._filter_traceback(item) for item in data]
+            data = [self._sanitize(item) for item in data]  # pyright: ignore[reportUnknownVariableType]
         else:
             return data
-        return data
+        return data  # pyright: ignore[reportUnknownVariableType]

src/pytest_broadcaster/_internal/_fields.py

@@ -1,11 +1,23 @@
 from __future__ import annotations
 
 import datetime
+import importlib.metadata
+import platform
+import sys
+from uuid import uuid4
 from warnings import WarningMessage
 
 import pytest
 from _pytest._code.code import ReprTraceback
 
+from pytest_broadcaster.models.python_distribution import (
+    Package,
+    Platform,
+    PythonDistribution,
+    Releaselevel,
+    Version,
+)
+
 from ._utils import (
     NodeID,
     TracebackLine,
@@ -19,6 +31,10 @@
 )
 
 
+def make_session_id() -> str:
+    return str(uuid4())
+
+
 def make_node_id(
     item: pytest.Item | pytest.Directory | pytest.Module | pytest.Class,
 ) -> NodeID:
@@ -59,6 +75,45 @@ def field_parameters(item: pytest.Item) -> dict[str, str]:
     return {k: type(v).__name__ for k, v in sorted(get_test_args(item).items())}
 
 
+def field_python() -> PythonDistribution:
+    packages = [
+        Package(name=x.metadata.get("Name"), version=x.version)
+        for x in importlib.metadata.distributions()
+    ]
+    platform_os = platform.system()
+    if platform_os == "Linux":
+        platform_os = Platform.linux
+    elif platform_os == "Darwin":
+        platform_os = Platform.darwin
+    elif platform_os == "Windows":
+        platform_os = Platform.windows
+    elif platform_os == "Java":
+        platform_os = Platform.java
+    else:
+        platform_os = Platform.unknown
+    processor_architecture = platform.processor()
+    level = sys.version_info.releaselevel
+    if level == "final":
+        level = Releaselevel.final
+    elif level == "beta":
+        level = Releaselevel.beta
+    elif level == "alpha":
+        level = Releaselevel.alpha
+    else:
+        level = Releaselevel.candidate
+    return PythonDistribution(
+        version=Version(
+            major=sys.version_info.major,
+            minor=sys.version_info.minor,
+            micro=sys.version_info.micro,
+            releaselevel=level,
+        ),
+        packages=packages,
+        platform=platform_os,
+        processor=processor_architecture,
+    )
+
+
 def make_warning_message(warning: WarningMessage) -> str:
     if isinstance(warning.message, str):
         return warning.message
@@ -69,6 +124,10 @@ def make_timestamp(epoch: float) -> str:
     return datetime.datetime.fromtimestamp(epoch).isoformat()
 
 
+def make_timestamp_from_datetime(dt: datetime.datetime) -> str:
+    return dt.astimezone(datetime.timezone.utc).isoformat()
+
+
 def make_traceback(report: pytest.TestReport) -> list[TracebackLine]:
     return make_traceback_from_reprtraceback(report.longrepr.reprtraceback)  # type: ignore