11"""Pytest-only hooks and fixtures for Module 01 tests."""
22
33import importlib
4- import os
54import sys
65from pathlib import Path
76
1716module_runner = importlib .import_module ("scripts.module_runner" )
1817
1918MODULE_DIR = module01_test_support .MODULE_DIR
20- ProcessManager = module01_test_support .ProcessManager
2119SECURITY_DIR = module01_test_support .SECURITY_DIR
22- _has_display = module01_test_support ._has_display
23- _security_artifacts_exist = module01_test_support ._security_artifacts_exist
24- _security_plugin_available = module01_test_support ._security_plugin_available
20+
21+ from scripts .test_utils import ( # noqa: E402
22+ ProcessManager ,
23+ UtilityApp ,
24+ has_display ,
25+ security_plugin_available ,
26+ )
27+
28+
29+ def _security_artifacts_exist () -> bool :
30+ """Return True when setup_security.py has been run for module 01."""
31+ domain_scope_dir = SECURITY_DIR / "domain_scope"
32+ if not domain_scope_dir .is_dir ():
33+ return False
34+ return any (domain_scope_dir .rglob ("*.p7s" ))
2535
2636
2737def pytest_collection_modifyitems (config , items ):
@@ -36,28 +46,31 @@ def pytest_collection_modifyitems(config, items):
3646 reason = "DDS Security runtime probe failed" ,
3747 )
3848
39- has_display = _has_display ()
49+ _has_display = has_display ()
50+ __has_security_plugin = security_plugin_available ()
51+ _has_security_artifacts = _security_artifacts_exist ()
52+
4053 module_items = [i for i in items if Path (i .fspath ).is_relative_to (TESTS_DIR )]
4154
4255 for item in module_items :
43- if "gui" in item .keywords and not has_display :
56+ if "gui" in item .keywords and not _has_display :
4457 item .add_marker (skip_gui )
45- if "secure" in item .keywords and not _security_artifacts_exist () :
58+ if "secure" in item .keywords and not _has_security_artifacts :
4659 item .add_marker (skip_sec_artifacts )
47- elif "secure" in item .keywords and not _security_plugin_available () :
60+ elif "secure" in item .keywords and not __has_security_plugin :
4861 item .add_marker (skip_sec_plugin )
4962
5063
5164@pytest .fixture (scope = "session" )
5265def dds_env ():
53- """Session-scoped environment dict with NDDS_QOS_PROFILES configured ( non-secure) ."""
66+ """Session-scoped environment for non-secure subprocess launches ."""
5467 env , apps = module_runner .load_module_config (MODULE_DIR , flags = {"security" : False })
5568 return env , apps
5669
5770
5871@pytest .fixture (scope = "session" )
5972def dds_env_secure ():
60- """Session-scoped environment dict with NDDS_QOS_PROFILES + DDS Security ."""
73+ """Session-scoped environment for secure subprocess launches ."""
6174 env , apps = module_runner .load_module_config (MODULE_DIR , flags = {"security" : True })
6275 env ["RTI_SECURITY_ARTIFACTS_DIR" ] = str (SECURITY_DIR )
6376 return env , apps
@@ -67,7 +80,7 @@ def dds_env_secure():
6780def proc_manager (dds_env ):
6881 """Yield a ProcessManager wired to the non-secure DDS environment."""
6982 env , apps = dds_env
70- pm = ProcessManager (env , apps )
83+ pm = ProcessManager (env , apps , cwd = MODULE_DIR )
7184 yield pm
7285 pm .shutdown_all ()
7386
@@ -76,7 +89,7 @@ def proc_manager(dds_env):
7689def class_proc_manager (dds_env ):
7790 """Class-scoped ProcessManager for read-only test classes."""
7891 env , apps = dds_env
79- pm = ProcessManager (env , apps )
92+ pm = ProcessManager (env , apps , cwd = MODULE_DIR )
8093 yield pm
8194 pm .shutdown_all ()
8295
@@ -85,21 +98,14 @@ def class_proc_manager(dds_env):
8598def proc_manager_secure (dds_env_secure ):
8699 """Yield a ProcessManager wired to the secure DDS environment."""
87100 env , apps = dds_env_secure
88- pm = ProcessManager (env , apps )
101+ pm = ProcessManager (env , apps , cwd = MODULE_DIR )
89102 yield pm
90103 pm .shutdown_all ()
91104
92105
93106@pytest .fixture ()
94- def dds_participant (dds_env ):
95- """Create a lightweight DDS DomainParticipant for test observation."""
96- import rti .connextdds as dds
97-
98- env , _apps = dds_env
99- os .environ ["NDDS_QOS_PROFILES" ] = env ["NDDS_QOS_PROFILES" ]
100-
101- provider = dds .QosProvider .default
102- participant_qos = provider .participant_qos_from_profile ("DpQosLib::Test" )
103- participant = dds .DomainParticipant (domain_id = 0 , qos = participant_qos )
104- yield participant
105- participant .close ()
107+ def nonsecure_utility_app ():
108+ """Create a UtilityApp for in-process DDS observation."""
109+ app = UtilityApp .make_non_secure ()
110+ yield app
111+ app .close ()
0 commit comments