18075468099

Committed 28 Sep 2025 02:13PM UTC coverage: 46.471% (+0.2%) from 46.252%

Build # 18075468099

Build Type

push

github

Committed by

djbrown

Commit Message

remove unused code from custom test runner

Run Details

48 of 222 branches covered (21.62%)

Branch coverage included in aggregate %.

0 of 1 new or added line in 1 file covered. (0.0%)

1697 of 3533 relevant lines covered (48.03%)

0.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

53.9

/src/base/tests/base.py

import os
import sys
from contextlib import contextmanager
from typing import Type

import pytest
from django.conf import settings
from django.contrib.staticfiles.testing import StaticLiveServerTestCase
from django.core.management import call_command
from django.core.servers.basehttp import ThreadedWSGIServer
from django.db.models import Model
from django.db.models.query import QuerySet
from django.test import TestCase, tag
from django.test.runner import DiscoverRunner
from django.test.testcases import LiveServerThread, QuietWSGIRequestHandler
from django.urls import ResolverMatch, resolve, reverse
from sauceclient import SauceClient
from selenium.webdriver import Firefox, Remote
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.safari.options import Options as SafariOptions
from selenium.webdriver.support.expected_conditions import staleness_of
from selenium.webdriver.support.ui import WebDriverWait


class Runner(DiscoverRunner):

    def __init__(self, verbosity=1, failfast=False, keepdb=False, **kwargs):
        self.verbosity = verbosity
        self.failfast = failfast
        self.keepdb = keepdb
        super().__init__(**kwargs)

    def build_suite(self, test_labels=None, **kwargs):
        suite = super().build_suite(test_labels, **kwargs)
        return filter_tests_by_explicit_tags(suite, self.tags)


def filter_tests_by_explicit_tags(suite, cmd_tags):
    suite_class = type(suite)
    filtered_suite = suite_class()

    for test in suite:
        # recursion
        if isinstance(test, suite_class):
            filtered_suite.addTests(filter_tests_by_explicit_tags(test, cmd_tags))
            continue

        # gather class and function tags
        class_tags = set(getattr(test, "explicit_tags", set()))
        func_name = getattr(test, "_testMethodName", str(test))
        func = getattr(test, func_name, test)
        func_tags = set(getattr(func, "explicit_tags", set()))
        code_tags = class_tags.union(func_tags)

        if should_include_test(cmd_tags, code_tags):
            filtered_suite.addTest(test)

    return filtered_suite


def should_include_test(cmd_tags, code_tags) -> bool:
    return not code_tags if not cmd_tags else bool(code_tags.intersection(cmd_tags))


def skip_unless_any_tag(*tags):
    base_decorator = tag(*tags)

    def decorator(obj):
        obj = base_decorator(obj)
        if hasattr(obj, "explicit_tags"):
            obj.tags = obj.tags.union(tags)
        else:
            setattr(obj, "explicit_tags", set(tags))
        return obj

    return decorator


class ModelTestCase(TestCase):

    def assert_objects(self, model: Type[Model], count=1, filters=None) -> Model | QuerySet[Model]:
        if filters is None:
            filters = {}

        objects = model._default_manager.filter(**filters)  # pylint: disable=protected-access
        self.assertEqual(len(objects), count)
        return objects[0] if count == 1 else objects


@pytest.mark.integration
@pytest.mark.slow
@skip_unless_any_tag("integration", "slow")
class IntegrationTestCase(ModelTestCase):

    def assert_command(self, command_name, *arguments, allow_error_logs=False, **options):
        if not allow_error_logs:
            with self.assertNoLogs("hbscorez", level="ERROR"):
                call_command(command_name, *arguments, **options)
        else:
            call_command(command_name, *arguments, **options)


class LiveServerThreadWithReuse(LiveServerThread):
    """
    This miniclass overrides _create_server to allow port reuse. This avoids creating
    "address already in use" errors for tests that have been run subsequently.
    """

    def _create_server(self, connections_override=None):
        return ThreadedWSGIServer(
            (self.host, self.port),
            QuietWSGIRequestHandler,
            allow_reuse_address=True,
            connections_override=connections_override,
        )


_CI = "CI" in os.environ
_SAUCE_BUILD = os.environ.get("SAUCE_BUILD_NAME")
_SAUCE_TUNNEL = os.environ.get("SAUCE_TUNNEL_IDENTIFIER")
_SAUCE_USER = os.environ.get("SAUCE_USERNAME")
_SAUCE_KEY = os.environ.get("SAUCE_ACCESS_KEY")


@pytest.mark.browser
@pytest.mark.slow
@skip_unless_any_tag("browser", "slow")
class BrowserTestCase(StaticLiveServerTestCase):

    port = 8001
    server_thread_class = LiveServerThreadWithReuse

    """Browser test cases are only run in CI or if configured explicitly"""

    def setUp(self):
        if _CI:
            self.driver = self.sauce_chrome_webdriver()
        else:
            options = FirefoxOptions()
            options.add_argument("-headless")
            self.driver = Firefox(options=options)
        self.driver.implicitly_wait(10)

    def sauce_chrome_webdriver(self):
        class_name = self.__class__.__name__
        method_name = self._testMethodName

        options = SafariOptions()
        options.browser_version = "14"
        options.platform_name = "macOS 11.00"
        sauce_options = {
            "name": f"{class_name}.{method_name}",
            "build": _SAUCE_BUILD,
            "tunnelIdentifier": _SAUCE_TUNNEL,
            "username": _SAUCE_USER,
            "accessKey": _SAUCE_KEY,
        }
        options.set_capability("sauce:options", sauce_options)

        remote_url = "https://ondemand.us-west-1.saucelabs.com:443/wd/hub"
        return Remote(command_executor=remote_url, options=options)

    def tearDown(self):
        self.driver.quit()
        if _CI:
            sauce_client = SauceClient(_SAUCE_USER, _SAUCE_KEY)
            status = sys.exc_info() == (None, None, None)
            sauce_client.jobs.update_job(job_id=self.driver.session_id, build=_SAUCE_TUNNEL, passed=status)

    def navigate(self, view_name: str, query: str = ""):
        path = reverse(view_name)
        self.driver.get(self.live_server_url + path + query)

    def assert_view(self, view_name: str):
        path: str = self.driver.current_url.replace(self.live_server_url, "").split("?")[0]
        resolved: ResolverMatch = resolve(path)
        self.assertEqual(resolved.view_name, view_name)

    @contextmanager
    def load(self, timeout=1):
        page = self.driver.find_element(By.TAG_NAME, "html")
        yield
        WebDriverWait(self.driver, timeout).until(staleness_of(page))

    @contextmanager
    def wait(self, timeout=settings.BROWSER_TIMEOUT):
        condition = _UrlHasChanged(self.driver.current_url)
        yield
        WebDriverWait(self.driver, timeout).until(condition)


class _UrlHasChanged:

    def __init__(self, url):
        self.old_url = url

    def __call__(self, driver):
        return driver.current_url != self.old_url

1	import os	1✔
2	import sys	1✔
3	from contextlib import contextmanager	1✔
4	from typing import Type	1✔
5
6	import pytest	1✔
7	from django.conf import settings	1✔
8	from django.contrib.staticfiles.testing import StaticLiveServerTestCase	1✔
9	from django.core.management import call_command	1✔
10	from django.core.servers.basehttp import ThreadedWSGIServer	1✔
11	from django.db.models import Model	1✔
12	from django.db.models.query import QuerySet	1✔
13	from django.test import TestCase, tag	1✔
14	from django.test.runner import DiscoverRunner	1✔
15	from django.test.testcases import LiveServerThread, QuietWSGIRequestHandler	1✔
16	from django.urls import ResolverMatch, resolve, reverse	1✔
17	from sauceclient import SauceClient	1✔
18	from selenium.webdriver import Firefox, Remote	1✔
19	from selenium.webdriver.common.by import By	1✔
20	from selenium.webdriver.firefox.options import Options as FirefoxOptions	1✔
21	from selenium.webdriver.safari.options import Options as SafariOptions	1✔
22	from selenium.webdriver.support.expected_conditions import staleness_of	1✔
23	from selenium.webdriver.support.ui import WebDriverWait	1✔
24
25
26	class Runner(DiscoverRunner):	1✔
27
28	def __init__(self, verbosity=1, failfast=False, keepdb=False, **kwargs):	1✔
29	self.verbosity = verbosity	×
30	self.failfast = failfast	×
31	self.keepdb = keepdb	×
32	super().__init__(**kwargs)	×
33
34	def build_suite(self, test_labels=None, **kwargs):	1✔
NEW 35	suite = super().build_suite(test_labels, **kwargs)	×
36	return filter_tests_by_explicit_tags(suite, self.tags)	×
37
38
39	def filter_tests_by_explicit_tags(suite, cmd_tags):	1✔
40	suite_class = type(suite)	×
41	filtered_suite = suite_class()	×
42
43	for test in suite:	×
44	# recursion
45	if isinstance(test, suite_class):	×
46	filtered_suite.addTests(filter_tests_by_explicit_tags(test, cmd_tags))	×
47	continue	×
48
49	# gather class and function tags
50	class_tags = set(getattr(test, "explicit_tags", set()))	×
51	func_name = getattr(test, "_testMethodName", str(test))	×
52	func = getattr(test, func_name, test)	×
53	func_tags = set(getattr(func, "explicit_tags", set()))	×
54	code_tags = class_tags.union(func_tags)	×
55
56	if should_include_test(cmd_tags, code_tags):	×
57	filtered_suite.addTest(test)	×
58
59	return filtered_suite	×
60
61
62	def should_include_test(cmd_tags, code_tags) -> bool:	1✔
63	return not code_tags if not cmd_tags else bool(code_tags.intersection(cmd_tags))	1✔
64
65
66	def skip_unless_any_tag(*tags):	1✔
67	base_decorator = tag(*tags)	1✔
68
69	def decorator(obj):	1✔
70	obj = base_decorator(obj)	1✔
71	if hasattr(obj, "explicit_tags"):	1✔
72	obj.tags = obj.tags.union(tags)	×
73	else:
74	setattr(obj, "explicit_tags", set(tags))	1✔
75	return obj	1✔
76
77	return decorator	1✔
78
79
80	class ModelTestCase(TestCase):	1✔
81
82	def assert_objects(self, model: Type[Model], count=1, filters=None) -> Model \| QuerySet[Model]:	1✔
83	if filters is None:	1✔
84	filters = {}	1✔
85
86	objects = model._default_manager.filter(**filters) # pylint: disable=protected-access	1✔
87	self.assertEqual(len(objects), count)	1✔
88	return objects[0] if count == 1 else objects	1✔
89
90
91	@pytest.mark.integration	1✔
92	@pytest.mark.slow	1✔
93	@skip_unless_any_tag("integration", "slow")	1✔
94	class IntegrationTestCase(ModelTestCase):	1✔
95
96	def assert_command(self, command_name, arguments, allow_error_logs=False, *options):	1✔
97	if not allow_error_logs:	×
98	with self.assertNoLogs("hbscorez", level="ERROR"):	×
99	call_command(command_name, arguments, *options)	×
100	else:
101	call_command(command_name, arguments, *options)	×
102
103
104	class LiveServerThreadWithReuse(LiveServerThread):	1✔
105	"""
106	This miniclass overrides _create_server to allow port reuse. This avoids creating
107	"address already in use" errors for tests that have been run subsequently.
108	"""
109
110	def _create_server(self, connections_override=None):	1✔
111	return ThreadedWSGIServer(	×
112	(self.host, self.port),
113	QuietWSGIRequestHandler,
114	allow_reuse_address=True,
115	connections_override=connections_override,
116	)
117
118
119	_CI = "CI" in os.environ	1✔
120	_SAUCE_BUILD = os.environ.get("SAUCE_BUILD_NAME")	1✔
121	_SAUCE_TUNNEL = os.environ.get("SAUCE_TUNNEL_IDENTIFIER")	1✔
122	_SAUCE_USER = os.environ.get("SAUCE_USERNAME")	1✔
123	_SAUCE_KEY = os.environ.get("SAUCE_ACCESS_KEY")	1✔
124
125
126	@pytest.mark.browser	1✔
127	@pytest.mark.slow	1✔
128	@skip_unless_any_tag("browser", "slow")	1✔
129	class BrowserTestCase(StaticLiveServerTestCase):	1✔
130
131	port = 8001	1✔
132	server_thread_class = LiveServerThreadWithReuse	1✔
133
134	"""Browser test cases are only run in CI or if configured explicitly"""	1✔
135
136	def setUp(self):	1✔
137	if _CI:	×
138	self.driver = self.sauce_chrome_webdriver()	×
139	else:
140	options = FirefoxOptions()	×
141	options.add_argument("-headless")	×
142	self.driver = Firefox(options=options)	×
143	self.driver.implicitly_wait(10)	×
144
145	def sauce_chrome_webdriver(self):	1✔
146	class_name = self.__class__.__name__	×
147	method_name = self._testMethodName	×
148
149	options = SafariOptions()	×
150	options.browser_version = "14"	×
151	options.platform_name = "macOS 11.00"	×
152	sauce_options = {	×
153	"name": f"{class_name}.{method_name}",
154	"build": _SAUCE_BUILD,
155	"tunnelIdentifier": _SAUCE_TUNNEL,
156	"username": _SAUCE_USER,
157	"accessKey": _SAUCE_KEY,
158	}
159	options.set_capability("sauce:options", sauce_options)	×
160
161	remote_url = "https://ondemand.us-west-1.saucelabs.com:443/wd/hub"	×
162	return Remote(command_executor=remote_url, options=options)	×
163
164	def tearDown(self):	1✔
165	self.driver.quit()	×
166	if _CI:	×
167	sauce_client = SauceClient(_SAUCE_USER, _SAUCE_KEY)	×
168	status = sys.exc_info() == (None, None, None)	×
169	sauce_client.jobs.update_job(job_id=self.driver.session_id, build=_SAUCE_TUNNEL, passed=status)	×
170
171	def navigate(self, view_name: str, query: str = ""):	1✔
172	path = reverse(view_name)	×
173	self.driver.get(self.live_server_url + path + query)	×
174
175	def assert_view(self, view_name: str):	1✔
176	path: str = self.driver.current_url.replace(self.live_server_url, "").split("?")[0]	×
177	resolved: ResolverMatch = resolve(path)	×
178	self.assertEqual(resolved.view_name, view_name)	×
179
180	@contextmanager	1✔
181	def load(self, timeout=1):	1✔
182	page = self.driver.find_element(By.TAG_NAME, "html")	×
183	yield	×
184	WebDriverWait(self.driver, timeout).until(staleness_of(page))	×
185
186	@contextmanager	1✔
187	def wait(self, timeout=settings.BROWSER_TIMEOUT):	1✔
188	condition = _UrlHasChanged(self.driver.current_url)	×
189	yield	×
190	WebDriverWait(self.driver, timeout).until(condition)	×
191
192
193	class _UrlHasChanged:	1✔
194
195	def __init__(self, url):	1✔
196	self.old_url = url	×
197
198	def __call__(self, driver):	1✔
199	return driver.current_url != self.old_url	×

djbrown / hbscorez / 18075468099

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous