123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144 |
- import contextlib
- import os
- import pytest
- pytest_plugins = ["sentry.utils.pytest"]
- @pytest.hookimpl(tryfirst=True, hookwrapper=True)
- def pytest_runtest_makereport(item, call):
-
- outcome = yield
- report = outcome.get_result()
-
-
- if os.environ.get("GITHUB_ACTIONS") != "true":
- return
-
-
-
- if hasattr(item, "execution_count"):
- import pytest_rerunfailures
- if item.execution_count <= pytest_rerunfailures.get_reruns_count(item):
- return
- if report.when == "call" and report.failed:
-
- filesystempath, lineno, _ = report.location
-
- workspace = os.environ.get("GITHUB_WORKSPACE")
- if workspace:
- full_path = os.path.abspath(filesystempath)
- try:
- rel_path = os.path.relpath(full_path, workspace)
- except ValueError:
-
-
-
- rel_path = filesystempath
- if not rel_path.startswith(".."):
- filesystempath = rel_path
- if lineno is not None:
-
- lineno += 1
-
- longrepr = report.head_line or item.name
-
- try:
- longrepr += "\n\n" + report.longrepr.reprcrash.message
- lineno = report.longrepr.reprcrash.lineno
- except AttributeError:
- pass
- print(_error_workflow_command(filesystempath, lineno, longrepr))
- def _error_workflow_command(filesystempath, lineno, longrepr):
-
- details_dict = {"file": filesystempath}
- if lineno is not None:
- details_dict["line"] = lineno
- details = ",".join(f"{k}={v}" for k, v in details_dict.items())
- if longrepr is None:
- return f"\n::error {details}"
- else:
- longrepr = _escape(longrepr)
- return f"\n::error {details}::{longrepr}"
- def _escape(s):
- return s.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A")
- _MODEL_MANIFEST_FILE_PATH = os.getenv("SENTRY_MODEL_MANIFEST_FILE_PATH")
- _model_manifest = None
- @pytest.fixture(scope="session", autouse=True)
- def create_model_manifest_file():
- """Audit which models are touched by each test case and write it to file."""
-
-
- from sentry.testutils.modelmanifest import ModelManifest
- if _MODEL_MANIFEST_FILE_PATH:
- global _model_manifest
- _model_manifest = ModelManifest.open(_MODEL_MANIFEST_FILE_PATH)
- with _model_manifest.write():
- yield
- else:
- yield
- @pytest.fixture(scope="class", autouse=True)
- def register_class_in_model_manifest(request: pytest.FixtureRequest):
- if _model_manifest:
- with _model_manifest.register(request.node.nodeid):
- yield
- else:
- yield
- @pytest.fixture(autouse=True)
- def setup_default_hybrid_cloud_stubs():
- from sentry.region_to_control.producer import (
- MockRegionToControlMessageService,
- region_to_control_message_service,
- )
- from sentry.silo import SiloMode
- from sentry.testutils.hybrid_cloud import service_stubbed
- stubs = [
- service_stubbed(
- region_to_control_message_service, MockRegionToControlMessageService(), SiloMode.REGION
- ),
- ]
- with contextlib.ExitStack() as stack:
- for stub in stubs:
- stack.enter_context(stub)
- yield
|