Skip to content

Python API

behave-trace can be used as a library for loading, inspecting, and serializing trace data.

Public API

import behave_trace
print(behave_trace.__version__)

Attachment helpers

behave_trace.attach_screenshot 

attach_screenshot(context: Any, source: Any, name: str = 'screenshot.png') -> None

Attach a screenshot to the current step.

source can be: bytes, path string, Selenium WebDriver, Playwright Page.

Source code in behave_trace/attach.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def attach_screenshot(context: Any, source: Any, name: str = "screenshot.png") -> None:
    """Attach a screenshot to the current step.

    source can be: bytes, path string, Selenium WebDriver, Playwright Page.
    """
    formatter = _find_formatter(context)
    if formatter is None:
        return

    data: bytes | None = None
    if isinstance(source, (bytes, bytearray)):
        data = bytes(source)
    elif isinstance(source, (str, Path)):
        try:
            data = Path(source).read_bytes()
        except Exception:
            return
    else:
        method = getattr(source, "get_screenshot_as_png", None)
        if callable(method):
            with contextlib.suppress(Exception):
                data = method()
        if data is None:
            method = getattr(source, "screenshot", None)
            if callable(method):
                try:
                    data = method()
                except Exception:
                    return

    if data is None:
        return

    formatter.attach(
        Artifact(
            type=ARTIFACT_SCREENSHOT,
            name=name,
            mime_type="image/png",
            data_base64=base64.b64encode(data).decode("ascii"),
        )
    )

behave_trace.attach_dom 

attach_dom(context: Any, source: Any, name: str = 'dom.html') -> None

Attach a DOM snapshot to the current step.

source can be: HTML string, Selenium WebDriver, Playwright Page.

When source is a WebDriver or Page, the current URL is extracted and injected as a <base> tag so relative URLs (CSS, JS, images) resolve correctly in the viewer's iframe.

Source code in behave_trace/attach.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def attach_dom(context: Any, source: Any, name: str = "dom.html") -> None:
    """Attach a DOM snapshot to the current step.

    source can be: HTML string, Selenium WebDriver, Playwright Page.

    When source is a WebDriver or Page, the current URL is extracted and
    injected as a ``<base>`` tag so relative URLs (CSS, JS, images) resolve
    correctly in the viewer's iframe.
    """
    formatter = _find_formatter(context)
    if formatter is None:
        return

    html: str | None = None
    base_url: str | None = None

    if isinstance(source, str):
        html = source
    else:
        # Try to get current URL for <base> tag injection
        current_url = getattr(source, "current_url", None)
        if current_url is None:
            current_url = getattr(source, "url", None)
        if isinstance(current_url, str):
            base_url = current_url

        try:
            page_source = getattr(source, "page_source", None)
        except Exception:
            page_source = None
        if page_source is not None:
            html = str(page_source)
        if html is None:
            method = getattr(source, "content", None)
            if callable(method):
                try:
                    html = method()
                except Exception:
                    return

    if html is None:
        return

    # Inject <base> tag so relative URLs resolve in the viewer's iframe
    if base_url and "<base " not in html:
        base_tag = f'<base href="{base_url}">'
        if "<head>" in html:
            html = html.replace("<head>", f"<head>{base_tag}", 1)
        elif "<head " in html:
            html = html.replace("<head ", f"<head>{base_tag} ", 1)
        else:
            html = base_tag + html

    formatter.attach(
        Artifact(
            type=ARTIFACT_DOM,
            name=name,
            mime_type="text/html",
            text=html,
        )
    )

behave_trace.attach_text 

attach_text(context: Any, text: str, name: str = 'note.txt') -> None

Attach a plain text snippet to the current step.

Source code in behave_trace/attach.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
def attach_text(context: Any, text: str, name: str = "note.txt") -> None:
    """Attach a plain text snippet to the current step."""
    formatter = _find_formatter(context)
    if formatter is None:
        return
    formatter.attach(
        Artifact(
            type=ARTIFACT_TEXT,
            name=name,
            mime_type="text/plain",
            text=str(text),
        )
    )

behave_trace.attach_network 

attach_network(context: Any, request_data: Any, name: str = 'network') -> None

Attach an HTTP request/response as a network artifact to the current step.

Parameters:

Name Type Description Default
context Any

The Behave context object.

 required
request_data Any

Can be: - A dict with keys: method, url, status, headers, body, response. - A Selenium request/response log entry (from driver.get_log). - A Playwright :class:Request or :class:Response object.

 required
name str

Artifact name (default: "network").

 'network'
Source code in behave_trace/attach.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def attach_network(context: Any, request_data: Any, name: str = "network") -> None:
    """Attach an HTTP request/response as a network artifact to the current step.

    Args:
        context: The Behave context object.
        request_data: Can be:
            - A dict with keys: ``method``, ``url``, ``status``, ``headers``,
              ``body``, ``response``.
            - A Selenium request/response log entry (from ``driver.get_log``).
            - A Playwright :class:`Request` or :class:`Response` object.
        name: Artifact name (default: "network").
    """
    formatter = _find_formatter(context)
    if formatter is None:
        return

    payload = _normalize_network_data(request_data)
    if payload is None:
        return

    import json as _json

    formatter.attach(
        Artifact(
            type=ARTIFACT_NETWORK,
            name=name,
            mime_type="application/json",
            text=_json.dumps(payload, default=str),
        )
    )

behave_trace.log 

log(context: Any, message: str, level: str = 'info') -> None

Append a log line to the current step.

Parameters:

Name Type Description Default
context Any

The Behave context object.

 required
message str

The log message text.

 required
level str

Log level — "info", "warning", or "error" (default: "info").

 'info'
Source code in behave_trace/attach.py 
161
162
163
164
165
166
167
168
169
170
171
172
def log(context: Any, message: str, level: str = "info") -> None:
    """Append a log line to the current step.

    Args:
        context: The Behave context object.
        message: The log message text.
        level: Log level — "info", "warning", or "error" (default: "info").
    """
    formatter = _find_formatter(context)
    if formatter is None:
        return
    formatter.log(str(message), level=level)

Trace model

behave_trace.models.Trace dataclass

Root of the trace tree — serialized to .trace file.

Source code in behave_trace/models.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
@dataclass(slots=True)
class Trace:
    """Root of the trace tree — serialized to .trace file."""

    version: str = "1"
    created_at: datetime = field(default_factory=datetime.now)
    features: list[Feature] = field(default_factory=list)
    environment: Environment = field(default_factory=Environment)
    stats: TraceStats = field(default_factory=TraceStats)

    @property
    def overall_status(self) -> str:
        statuses = [f.status for f in self.features]
        if any(s == STATUS_FAILED for s in statuses):
            return STATUS_FAILED
        if any(s == STATUS_UNDEFINED for s in statuses):
            return STATUS_UNDEFINED
        if any(s == STATUS_PASSED for s in statuses):
            return STATUS_PASSED
        if statuses and all(s == STATUS_SKIPPED for s in statuses):
            return STATUS_SKIPPED
        return STATUS_UNTESTED

    def to_dict(self) -> dict[str, Any]:
        """Serialize including computed properties for the viewer frontend."""
        return {
            "version": self.version,
            "created_at": self.created_at.isoformat(),
            "features": [f.to_dict() for f in self.features],
            "environment": as_dict(self.environment),
            "stats": as_dict(self.stats),
            "overall_status": self.overall_status,
        }

to_dict 

to_dict() -> dict[str, Any]

Serialize including computed properties for the viewer frontend.

Source code in behave_trace/models.py 
408
409
410
411
412
413
414
415
416
417
def to_dict(self) -> dict[str, Any]:
    """Serialize including computed properties for the viewer frontend."""
    return {
        "version": self.version,
        "created_at": self.created_at.isoformat(),
        "features": [f.to_dict() for f in self.features],
        "environment": as_dict(self.environment),
        "stats": as_dict(self.stats),
        "overall_status": self.overall_status,
    }

behave_trace.models.Feature dataclass

A Gherkin feature.

Source code in behave_trace/models.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
@dataclass(slots=True)
class Feature:
    """A Gherkin feature."""

    name: str
    status: str = STATUS_UNTESTED
    duration: float = 0.0
    description: str = ""
    location: str = ""
    tags: list[str] = field(default_factory=list)
    scenarios: list[Scenario] = field(default_factory=list)
    background: Background | None = None

    @property
    def scenario_count(self) -> int:
        return len(self.scenarios)

    def to_dict(self) -> dict[str, Any]:
        """Serialize including computed properties for the viewer frontend."""
        return {
            "name": self.name,
            "status": self.status,
            "duration": self.duration,
            "description": self.description,
            "location": self.location,
            "tags": self.tags,
            "scenarios": [s.to_dict() for s in self.scenarios],
            "background": as_dict(self.background) if self.background else None,
            "scenario_count": self.scenario_count,
        }

to_dict 

to_dict() -> dict[str, Any]

Serialize including computed properties for the viewer frontend.

Source code in behave_trace/models.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
def to_dict(self) -> dict[str, Any]:
    """Serialize including computed properties for the viewer frontend."""
    return {
        "name": self.name,
        "status": self.status,
        "duration": self.duration,
        "description": self.description,
        "location": self.location,
        "tags": self.tags,
        "scenarios": [s.to_dict() for s in self.scenarios],
        "background": as_dict(self.background) if self.background else None,
        "scenario_count": self.scenario_count,
    }

behave_trace.models.Scenario dataclass

A scenario or scenario outline example.

Source code in behave_trace/models.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@dataclass(slots=True)
class Scenario:
    """A scenario or scenario outline example."""

    name: str
    status: str = STATUS_UNTESTED
    duration: float = 0.0
    description: str = ""
    location: str = ""
    tags: list[str] = field(default_factory=list)
    steps: list[Step] = field(default_factory=list)
    background: Background | None = None
    feature_name: str = ""
    rule_name: str = ""
    is_outline: bool = False
    outline_name: str = ""
    examples: DataTable | None = None

    @property
    def step_count(self) -> int:
        return len(self.steps)

    @property
    def passed_steps(self) -> int:
        return sum(1 for s in self.steps if s.status == STATUS_PASSED)

    @property
    def failed_steps(self) -> int:
        return sum(1 for s in self.steps if s.status == STATUS_FAILED)

    def to_dict(self) -> dict[str, Any]:
        """Serialize including computed properties for the viewer frontend."""
        return {
            "name": self.name,
            "status": self.status,
            "duration": self.duration,
            "description": self.description,
            "location": self.location,
            "tags": self.tags,
            "steps": [s.to_dict() for s in self.steps],
            "background": as_dict(self.background) if self.background else None,
            "feature_name": self.feature_name,
            "rule_name": self.rule_name,
            "is_outline": self.is_outline,
            "outline_name": self.outline_name,
            "examples": as_dict(self.examples) if self.examples else None,
            "step_count": self.step_count,
            "passed_steps": self.passed_steps,
            "failed_steps": self.failed_steps,
        }

to_dict 

to_dict() -> dict[str, Any]

Serialize including computed properties for the viewer frontend.

Source code in behave_trace/models.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def to_dict(self) -> dict[str, Any]:
    """Serialize including computed properties for the viewer frontend."""
    return {
        "name": self.name,
        "status": self.status,
        "duration": self.duration,
        "description": self.description,
        "location": self.location,
        "tags": self.tags,
        "steps": [s.to_dict() for s in self.steps],
        "background": as_dict(self.background) if self.background else None,
        "feature_name": self.feature_name,
        "rule_name": self.rule_name,
        "is_outline": self.is_outline,
        "outline_name": self.outline_name,
        "examples": as_dict(self.examples) if self.examples else None,
        "step_count": self.step_count,
        "passed_steps": self.passed_steps,
        "failed_steps": self.failed_steps,
    }

behave_trace.models.Step dataclass

A single Gherkin step with its execution trace.

Source code in behave_trace/models.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
@dataclass(slots=True)
class Step:
    """A single Gherkin step with its execution trace."""

    keyword: str
    name: str
    status: str = STATUS_UNTESTED
    duration: float = 0.0
    location: str = ""
    text: str | None = None
    table: DataTable | None = None
    error: ErrorInfo | None = None
    artifacts: list[Artifact] = field(default_factory=list)
    logs: list[str | dict[str, Any]] = field(default_factory=list)

    @property
    def has_screenshot(self) -> bool:
        return any(a.type == ARTIFACT_SCREENSHOT for a in self.artifacts)

    @property
    def has_dom(self) -> bool:
        return any(a.type == ARTIFACT_DOM for a in self.artifacts)

    @property
    def has_network(self) -> bool:
        return any(a.type == ARTIFACT_NETWORK for a in self.artifacts)

    def to_dict(self) -> dict[str, Any]:
        """Serialize including computed properties for the viewer frontend."""
        return {
            "keyword": self.keyword,
            "name": self.name,
            "status": self.status,
            "duration": self.duration,
            "location": self.location,
            "text": self.text,
            "table": as_dict(self.table) if self.table else None,
            "error": as_dict(self.error) if self.error else None,
            "artifacts": [as_dict(a) for a in self.artifacts],
            "logs": self.logs,
            "has_screenshot": self.has_screenshot,
            "has_dom": self.has_dom,
            "has_network": self.has_network,
        }

to_dict 

to_dict() -> dict[str, Any]

Serialize including computed properties for the viewer frontend.

Source code in behave_trace/models.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def to_dict(self) -> dict[str, Any]:
    """Serialize including computed properties for the viewer frontend."""
    return {
        "keyword": self.keyword,
        "name": self.name,
        "status": self.status,
        "duration": self.duration,
        "location": self.location,
        "text": self.text,
        "table": as_dict(self.table) if self.table else None,
        "error": as_dict(self.error) if self.error else None,
        "artifacts": [as_dict(a) for a in self.artifacts],
        "logs": self.logs,
        "has_screenshot": self.has_screenshot,
        "has_dom": self.has_dom,
        "has_network": self.has_network,
    }

behave_trace.models.Artifact dataclass

An artifact captured during a step execution.

Source code in behave_trace/models.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@dataclass(slots=True)
class Artifact:
    """An artifact captured during a step execution."""

    type: str
    name: str = ""
    mime_type: str = "application/octet-stream"
    data_base64: str = ""
    text: str | None = None

    @property
    def is_image(self) -> bool:
        return self.mime_type.startswith("image/")

    @property
    def is_text(self) -> bool:
        return self.mime_type.startswith("text/") or self.mime_type in {
            "application/json",
            "application/xml",
            "text/html",
        }

    def to_dict(self) -> dict[str, Any]:
        """Serialize including computed properties for the viewer frontend."""
        return {
            "type": self.type,
            "name": self.name,
            "mime_type": self.mime_type,
            "data_base64": self.data_base64,
            "text": self.text,
        }

to_dict 

to_dict() -> dict[str, Any]

Serialize including computed properties for the viewer frontend.

Source code in behave_trace/models.py 
114
115
116
117
118
119
120
121
122
def to_dict(self) -> dict[str, Any]:
    """Serialize including computed properties for the viewer frontend."""
    return {
        "type": self.type,
        "name": self.name,
        "mime_type": self.mime_type,
        "data_base64": self.data_base64,
        "text": self.text,
    }

Serializer

behave_trace.serializer.Serializer

Serialize and deserialize Trace objects.

Source code in behave_trace/serializer.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class Serializer:
    """Serialize and deserialize Trace objects."""

    @staticmethod
    def save(trace: Trace, path: str | Path) -> Path:
        """Save a trace to a JSON file.

        Args:
            trace: The Trace object to save.
            path: Output file path.

        Returns:
            The path where the trace was written.
        """
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        data = as_dict(trace)
        p.write_text(json.dumps(data, indent=2, default=str, ensure_ascii=False), encoding="utf-8")
        return p

    @staticmethod
    def load(path: str | Path) -> Trace:
        """Load a trace from a JSON file.

        Args:
            path: Input file path.

        Returns:
            Reconstructed Trace object.

        Raises:
            FileNotFoundError: If the file does not exist.
            json.JSONDecodeError: If the file is not valid JSON.
            ValueError: If the JSON root is not a JSON object.
        """
        p = Path(path)
        if not p.exists():
            raise FileNotFoundError(f"Trace file not found: {p}")
        data = json.loads(p.read_text(encoding="utf-8"))
        if not isinstance(data, dict):
            raise ValueError(f"Expected JSON object at root, got {type(data).__name__}")
        return Serializer._from_dict(data)

    @staticmethod
    def _from_dict(data: dict[str, Any]) -> Trace:
        """Reconstruct a Trace from a plain dict."""
        trace = Trace(
            version=data.get("version") or "1",
            features=[],
        )
        created = data.get("created_at")
        if isinstance(created, str):
            with contextlib.suppress(ValueError):
                trace.created_at = datetime.fromisoformat(created)
        for f_data in _as_list(data.get("features")):
            if not isinstance(f_data, dict):
                continue
            feature = _feature_from_dict(f_data)
            trace.features.append(feature)
        env_data = _as_dict(data.get("environment"))
        trace.environment = _environment_from_dict(env_data)
        stats_data = _as_dict(data.get("stats"))
        trace.stats = _stats_from_dict(stats_data)
        return trace

load staticmethod 

load(path: str | Path) -> Trace

Load a trace from a JSON file.

Parameters:

Name Type Description Default
path str | Path

Input file path.

 required

Returns:

Type Description
Trace

Reconstructed Trace object.

Raises:

Type Description
FileNotFoundError

If the file does not exist.
JSONDecodeError

If the file is not valid JSON.
ValueError

If the JSON root is not a JSON object.
Source code in behave_trace/serializer.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@staticmethod
def load(path: str | Path) -> Trace:
    """Load a trace from a JSON file.

    Args:
        path: Input file path.

    Returns:
        Reconstructed Trace object.

    Raises:
        FileNotFoundError: If the file does not exist.
        json.JSONDecodeError: If the file is not valid JSON.
        ValueError: If the JSON root is not a JSON object.
    """
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"Trace file not found: {p}")
    data = json.loads(p.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ValueError(f"Expected JSON object at root, got {type(data).__name__}")
    return Serializer._from_dict(data)

save staticmethod 

save(trace: Trace, path: str | Path) -> Path

Save a trace to a JSON file.

Parameters:

Name Type Description Default
trace Trace

The Trace object to save.

 required
path str | Path

Output file path.

 required

Returns:

Type Description
Path

The path where the trace was written.
Source code in behave_trace/serializer.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@staticmethod
def save(trace: Trace, path: str | Path) -> Path:
    """Save a trace to a JSON file.

    Args:
        trace: The Trace object to save.
        path: Output file path.

    Returns:
        The path where the trace was written.
    """
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    data = as_dict(trace)
    p.write_text(json.dumps(data, indent=2, default=str, ensure_ascii=False), encoding="utf-8")
    return p

Runner

behave_trace.runner.BehaveRunner

Execute behave as a subprocess and load the resulting trace.

Parameters:

Name Type Description Default
behave_executable str | Path | None

Path to the behave executable. Defaults to behave found on PATH (or python -m behave).

 None
Source code in behave_trace/runner.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class BehaveRunner:
    """Execute behave as a subprocess and load the resulting trace.

    Args:
        behave_executable: Path to the behave executable.
            Defaults to ``behave`` found on PATH (or ``python -m behave``).
    """

    def __init__(self, behave_executable: str | Path | None = None) -> None:
        if behave_executable:
            self._behave = str(behave_executable)
            self._use_module = False
        else:
            # Always use `python -m behave` to guarantee the same interpreter
            # and make behave_trace importable for the formatter.
            self._behave = sys.executable
            self._use_module = True

    def build_command(
        self,
        features_dir: str | Path = ".",
        output_path: str | Path = "trace.json",
        tags: str | None = None,
        extra_args: list[str] | None = None,
    ) -> list[str]:
        """Build the behave command line without executing it.

        Args:
            features_dir: Directory containing .feature files.
            output_path: Where the trace JSON will be written.
            tags: Optional tag expression (e.g. ``@smoke``).
            extra_args: Additional arguments to pass to behave.

        Returns:
            The command list suitable for :func:`subprocess.run`.
        """
        cmd = [sys.executable, "-m", "behave"] if self._use_module else [self._behave]

        cmd.extend(["--format", "behave-trace", "-o", str(output_path)])
        if tags:
            cmd.extend(["--tags", tags])
        if extra_args:
            cmd.extend(extra_args)
        cmd.append(str(features_dir))
        return cmd

    def run(
        self,
        features_dir: str | Path = ".",
        output_path: str | Path = "trace.json",
        tags: str | None = None,
        extra_args: list[str] | None = None,
        cwd: str | Path | None = None,
        server_url: str | None = None,
    ) -> RunResult:
        """Execute behave and return the result.

        Args:
            features_dir: Directory containing .feature files.
            output_path: Where the trace JSON will be written.
            tags: Optional tag expression.
            extra_args: Additional arguments for behave.
            cwd: Working directory for the subprocess.

        Returns:
            :class:`RunResult` with exit code, output, and trace path.
        """
        cmd = self.build_command(features_dir, output_path, tags, extra_args)

        # Ensure behave_trace is importable in the subprocess. When the package
        # is not installed (e.g. running from source), we inject the project
        # root into PYTHONPATH so the formatter registration in behave.ini works.
        env = None
        candidate_root = Path(__file__).resolve().parent.parent
        if (candidate_root / "behave_trace" / "__init__.py").exists():
            import os

            env = os.environ.copy()
            existing = env.get("PYTHONPATH", "")
            if existing:
                env["PYTHONPATH"] = f"{candidate_root}{os.pathsep}{existing}"
            else:
                env["PYTHONPATH"] = str(candidate_root)

        if server_url:
            env = env or os.environ.copy()
            env["BEHAVE_TRACE_SERVER_URL"] = str(server_url)

        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            cwd=str(cwd) if cwd else None,
            env=env,
        )
        trace_path = Path(output_path)
        if cwd:
            trace_path = Path(cwd) / trace_path
        return RunResult(
            returncode=result.returncode,
            stdout=result.stdout,
            stderr=result.stderr,
            trace_path=trace_path if trace_path.exists() else None,
        )

    def run_and_load(
        self,
        features_dir: str | Path = ".",
        output_path: str | Path = "trace.json",
        tags: str | None = None,
        extra_args: list[str] | None = None,
        cwd: str | Path | None = None,
    ) -> tuple[RunResult, Trace | None]:
        """Execute behave and load the resulting trace.

        Returns a tuple of (RunResult, Trace | None).
        If behave fails to produce a trace file, the second element is None.
        """

        result = self.run(
            features_dir=features_dir,
            output_path=output_path,
            tags=tags,
            extra_args=extra_args,
            cwd=cwd,
        )
        if result.trace_path and result.trace_path.exists():
            try:
                trace = Serializer.load(result.trace_path)
                return result, trace
            except Exception:
                return result, None
        return result, None

    def run_filtered(
        self,
        features_dir: str | Path = ".",
        output_path: str | Path = "trace.json",
        tags: str | None = None,
        scenario_names: list[str] | None = None,
        cwd: str | Path | None = None,
        server_url: str | None = None,
    ) -> RunResult:
        """Execute behave filtered by scenario names.

        Uses ``--name`` flags to select specific scenarios. If
        ``scenario_names`` is None or empty, behaves like :meth:`run`.

        Args:
            features_dir: Directory containing .feature files.
            output_path: Where the trace JSON will be written.
            tags: Optional tag expression.
            scenario_names: List of scenario names to run.
            cwd: Working directory for the subprocess.

        Returns:
            :class:`RunResult` with exit code, output, and trace path.
        """
        extra_args: list[str] = []
        if scenario_names:
            for name in scenario_names:
                extra_args.extend(["--name", name])
        return self.run(
            features_dir=features_dir,
            output_path=output_path,
            tags=tags,
            extra_args=extra_args,
            cwd=cwd,
            server_url=server_url,
        )

build_command 

build_command(features_dir: str | Path = '.', output_path: str | Path = 'trace.json', tags: str | None = None, extra_args: list[str] | None = None) -> list[str]

Build the behave command line without executing it.

Parameters:

Name Type Description Default
features_dir str | Path

Directory containing .feature files.

 '.'
output_path str | Path

Where the trace JSON will be written.

 'trace.json'
tags str | None

Optional tag expression (e.g. @smoke).

 None
extra_args list[str] | None

Additional arguments to pass to behave.

 None

Returns:

Type Description
list[str]

The command list suitable for :func:subprocess.run.
Source code in behave_trace/runner.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def build_command(
    self,
    features_dir: str | Path = ".",
    output_path: str | Path = "trace.json",
    tags: str | None = None,
    extra_args: list[str] | None = None,
) -> list[str]:
    """Build the behave command line without executing it.

    Args:
        features_dir: Directory containing .feature files.
        output_path: Where the trace JSON will be written.
        tags: Optional tag expression (e.g. ``@smoke``).
        extra_args: Additional arguments to pass to behave.

    Returns:
        The command list suitable for :func:`subprocess.run`.
    """
    cmd = [sys.executable, "-m", "behave"] if self._use_module else [self._behave]

    cmd.extend(["--format", "behave-trace", "-o", str(output_path)])
    if tags:
        cmd.extend(["--tags", tags])
    if extra_args:
        cmd.extend(extra_args)
    cmd.append(str(features_dir))
    return cmd

run 

run(features_dir: str | Path = '.', output_path: str | Path = 'trace.json', tags: str | None = None, extra_args: list[str] | None = None, cwd: str | Path | None = None, server_url: str | None = None) -> RunResult

Execute behave and return the result.

Parameters:

Name Type Description Default
features_dir str | Path

Directory containing .feature files.

 '.'
output_path str | Path

Where the trace JSON will be written.

 'trace.json'
tags str | None

Optional tag expression.

 None
extra_args list[str] | None

Additional arguments for behave.

 None
cwd str | Path | None

Working directory for the subprocess.

 None

Returns:

Type Description
RunResult

class:RunResult with exit code, output, and trace path.
Source code in behave_trace/runner.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def run(
    self,
    features_dir: str | Path = ".",
    output_path: str | Path = "trace.json",
    tags: str | None = None,
    extra_args: list[str] | None = None,
    cwd: str | Path | None = None,
    server_url: str | None = None,
) -> RunResult:
    """Execute behave and return the result.

    Args:
        features_dir: Directory containing .feature files.
        output_path: Where the trace JSON will be written.
        tags: Optional tag expression.
        extra_args: Additional arguments for behave.
        cwd: Working directory for the subprocess.

    Returns:
        :class:`RunResult` with exit code, output, and trace path.
    """
    cmd = self.build_command(features_dir, output_path, tags, extra_args)

    # Ensure behave_trace is importable in the subprocess. When the package
    # is not installed (e.g. running from source), we inject the project
    # root into PYTHONPATH so the formatter registration in behave.ini works.
    env = None
    candidate_root = Path(__file__).resolve().parent.parent
    if (candidate_root / "behave_trace" / "__init__.py").exists():
        import os

        env = os.environ.copy()
        existing = env.get("PYTHONPATH", "")
        if existing:
            env["PYTHONPATH"] = f"{candidate_root}{os.pathsep}{existing}"
        else:
            env["PYTHONPATH"] = str(candidate_root)

    if server_url:
        env = env or os.environ.copy()
        env["BEHAVE_TRACE_SERVER_URL"] = str(server_url)

    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        cwd=str(cwd) if cwd else None,
        env=env,
    )
    trace_path = Path(output_path)
    if cwd:
        trace_path = Path(cwd) / trace_path
    return RunResult(
        returncode=result.returncode,
        stdout=result.stdout,
        stderr=result.stderr,
        trace_path=trace_path if trace_path.exists() else None,
    )

run_and_load 

run_and_load(features_dir: str | Path = '.', output_path: str | Path = 'trace.json', tags: str | None = None, extra_args: list[str] | None = None, cwd: str | Path | None = None) -> tuple[RunResult, Trace | None]

Execute behave and load the resulting trace.

Returns a tuple of (RunResult, Trace | None). If behave fails to produce a trace file, the second element is None.

Source code in behave_trace/runner.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def run_and_load(
    self,
    features_dir: str | Path = ".",
    output_path: str | Path = "trace.json",
    tags: str | None = None,
    extra_args: list[str] | None = None,
    cwd: str | Path | None = None,
) -> tuple[RunResult, Trace | None]:
    """Execute behave and load the resulting trace.

    Returns a tuple of (RunResult, Trace | None).
    If behave fails to produce a trace file, the second element is None.
    """

    result = self.run(
        features_dir=features_dir,
        output_path=output_path,
        tags=tags,
        extra_args=extra_args,
        cwd=cwd,
    )
    if result.trace_path and result.trace_path.exists():
        try:
            trace = Serializer.load(result.trace_path)
            return result, trace
        except Exception:
            return result, None
    return result, None

run_filtered 

run_filtered(features_dir: str | Path = '.', output_path: str | Path = 'trace.json', tags: str | None = None, scenario_names: list[str] | None = None, cwd: str | Path | None = None, server_url: str | None = None) -> RunResult

Execute behave filtered by scenario names.

Uses --name flags to select specific scenarios. If scenario_names is None or empty, behaves like :meth:run.

Parameters:

Name Type Description Default
features_dir str | Path

Directory containing .feature files.

 '.'
output_path str | Path

Where the trace JSON will be written.

 'trace.json'
tags str | None

Optional tag expression.

 None
scenario_names list[str] | None

List of scenario names to run.

 None
cwd str | Path | None

Working directory for the subprocess.

 None

Returns:

Type Description
RunResult

class:RunResult with exit code, output, and trace path.
Source code in behave_trace/runner.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def run_filtered(
    self,
    features_dir: str | Path = ".",
    output_path: str | Path = "trace.json",
    tags: str | None = None,
    scenario_names: list[str] | None = None,
    cwd: str | Path | None = None,
    server_url: str | None = None,
) -> RunResult:
    """Execute behave filtered by scenario names.

    Uses ``--name`` flags to select specific scenarios. If
    ``scenario_names`` is None or empty, behaves like :meth:`run`.

    Args:
        features_dir: Directory containing .feature files.
        output_path: Where the trace JSON will be written.
        tags: Optional tag expression.
        scenario_names: List of scenario names to run.
        cwd: Working directory for the subprocess.

    Returns:
        :class:`RunResult` with exit code, output, and trace path.
    """
    extra_args: list[str] = []
    if scenario_names:
        for name in scenario_names:
            extra_args.extend(["--name", name])
    return self.run(
        features_dir=features_dir,
        output_path=output_path,
        tags=tags,
        extra_args=extra_args,
        cwd=cwd,
        server_url=server_url,
    )

behave_trace.runner.RunResult dataclass

Outcome of a behave run.

Source code in behave_trace/runner.py 
21
22
23
24
25
26
27
28
@dataclass(slots=True)
class RunResult:
    """Outcome of a behave run."""

    returncode: int
    stdout: str
    stderr: str
    trace_path: Path | None = None

Viewer

behave_trace.viewer.server.ViewerServer

Serve the trace viewer SPA and trace data on localhost.

Parameters:

Name Type Description Default
trace Trace | None

The :class:Trace object to visualize.

 None
port int

Port to bind (0 = auto-select a free port).

 0
watching bool

Whether the server is in watch mode (enables SSE and run-status endpoints).

 False
rerun_callback Callable[[list[str] | None], None] | None

Optional callback invoked when the client POSTs to /api/rerun. Receives a list of scenario names or None.

 None
Example

server = ViewerServer(trace, port=8080) server.start() 'http://127.0.0.1:8080' server.stop()

Source code in behave_trace/viewer/server.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
class ViewerServer:
    """Serve the trace viewer SPA and trace data on localhost.

    Args:
        trace: The :class:`Trace` object to visualize.
        port: Port to bind (0 = auto-select a free port).
        watching: Whether the server is in watch mode (enables SSE and
            run-status endpoints).
        rerun_callback: Optional callback invoked when the client POSTs
            to ``/api/rerun``. Receives a list of scenario names or None.

    Example:
        >>> server = ViewerServer(trace, port=8080)
        >>> server.start()
        'http://127.0.0.1:8080'
        >>> server.stop()
    """

    def __init__(
        self,
        trace: Trace | None = None,
        port: int = 0,
        watching: bool = False,
        rerun_callback: Callable[[list[str] | None], None] | None = None,
    ) -> None:
        if trace is None:
            from behave_trace.models import Trace as _Trace

            trace = _Trace()
        self.trace = trace
        self.port = port
        self.watching = watching
        self.rerun_callback = rerun_callback
        self._httpd: ThreadingHTTPServer | None = None
        self._thread: threading.Thread | None = None
        self._trace_json: bytes = json.dumps(as_dict(self.trace), default=str).encode()
        self._base_dir: Path = Path(trace.environment.cwd or ".").resolve()
        self._state = _ServerState(
            self._trace_json,
            watching,
            can_run=rerun_callback is not None,
            auto_run=watching,
        )

    @property
    def url(self) -> str:
        """Return the URL the server is listening on."""
        actual_port = self._httpd.server_address[1] if self._httpd is not None else self.port
        return f"http://127.0.0.1:{actual_port}"

    def start(self) -> str:
        """Start the server and return the URL."""
        state = self._state
        base_dir = self._base_dir
        rerun_cb = self.rerun_callback

        class Handler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def do_GET(self) -> None:
                parsed = urlparse(self.path)
                if parsed.path == "/api/trace":
                    self._send_json(state.trace_bytes, state.trace_gzipped)
                elif parsed.path == "/api/source":
                    self._serve_source(parsed.query, base_dir)
                elif parsed.path == "/api/watching":
                    payload = json.dumps({"watching": state.watching}).encode()
                    self._send_json(payload, gzip.compress(payload))
                elif parsed.path == "/api/stream":
                    self._handle_sse(state)
                elif parsed.path == "/":
                    self._serve_file(_ASSETS_DIR / "index.html")
                elif parsed.path.startswith("/css/") or parsed.path.startswith("/js/"):
                    self._serve_path(parsed.path.lstrip("/"))
                else:
                    self.send_error(404)

            def do_POST(self) -> None:
                parsed = urlparse(self.path)
                if parsed.path == "/api/rerun":
                    self._handle_rerun(state, rerun_cb)
                elif parsed.path == "/api/run":
                    self._handle_run(state, rerun_cb)
                elif parsed.path == "/api/autorun":
                    self._handle_autorun(state)
                elif parsed.path == "/api/progress":
                    self._handle_progress(state)
                else:
                    self._send_json_response({"error": "Not found"}, status=404)

            def _handle_progress(self, sstate: _ServerState) -> None:
                """Handle POST /api/progress — update live progress."""
                try:
                    content_length = int(self.headers.get("Content-Length", 0))
                except (ValueError, TypeError):
                    content_length = 0
                body = self.rfile.read(content_length) if content_length > 0 else b"{}"
                try:
                    payload = json.loads(body)
                except json.JSONDecodeError:
                    self._send_json_response({"error": "Invalid JSON"}, status=400)
                    return

                if not isinstance(payload, dict):
                    self._send_json_response({"error": "Expected JSON object"}, status=400)
                    return

                completed = payload.get("completed", sstate.progress["completed"])
                total = payload.get("total", sstate.progress["total"])
                sstate.progress = {
                    "completed": int(completed),
                    "total": int(total),
                }
                sstate.notify(
                    {
                        "type": "scenario_completed",
                        "completed": sstate.progress["completed"],
                        "total": sstate.progress["total"],
                        "scenario_name": payload.get("scenario_name", ""),
                    }
                )
                self._send_json_response({"status": "ok"})

            def _handle_autorun(self, sstate: _ServerState) -> None:
                """Handle POST /api/autorun — toggle auto-run on/off."""
                try:
                    content_length = int(self.headers.get("Content-Length", 0))
                except (ValueError, TypeError):
                    content_length = 0
                body = self.rfile.read(content_length) if content_length > 0 else b"{}"
                try:
                    payload = json.loads(body)
                except json.JSONDecodeError:
                    self._send_json_response({"error": "Invalid JSON"}, status=400)
                    return

                if not isinstance(payload, dict):
                    self._send_json_response({"error": "Expected JSON object"}, status=400)
                    return

                enabled = payload.get("enabled")
                if not isinstance(enabled, bool):
                    self._send_json_response({"error": "Missing 'enabled' boolean"}, status=400)
                    return

                sstate.auto_run = enabled
                sstate.notify({"type": "state", "autoRun": enabled})
                self._send_json_response({"status": "ok", "autoRun": enabled})

            def _handle_run(
                self,
                sstate: _ServerState,
                cb: Callable[[list[str] | None], None] | None,
            ) -> None:
                """Handle POST /api/run — execute behave from scratch (all tests)."""
                if cb is None:
                    self._send_json_response(
                        {"error": "Run not available (no callback configured)"},
                        status=501,
                    )
                    return
                if sstate.running:
                    self._send_json_response({"error": "Already running"}, status=409)
                    return
                self._send_json_response({"status": "accepted"})
                thread = threading.Thread(target=cb, args=(None,), daemon=True)
                thread.start()

            def _handle_rerun(
                self,
                sstate: _ServerState,
                cb: Callable[[list[str] | None], None] | None,
            ) -> None:
                """Handle POST /api/rerun — re-execute behave with optional filter."""
                if cb is None:
                    self._send_json_response(
                        {"error": "Re-run not available (no callback configured)"},
                        status=501,
                    )
                    return

                try:
                    content_length = int(self.headers.get("Content-Length", 0))
                except (ValueError, TypeError):
                    content_length = 0
                body = self.rfile.read(content_length) if content_length > 0 else b"{}"
                try:
                    payload = json.loads(body)
                except json.JSONDecodeError:
                    self._send_json_response({"error": "Invalid JSON"}, status=400)
                    return

                if not isinstance(payload, dict):
                    self._send_json_response({"error": "Expected JSON object"}, status=400)
                    return

                filter_type = payload.get("filter", "all")
                scenarios = payload.get("scenarios")

                if filter_type not in ("failed", "all"):
                    self._send_json_response(
                        {"error": "Invalid filter; must be 'failed' or 'all'"},
                        status=400,
                    )
                    return

                scenario_names: list[str] | None = None
                if filter_type == "failed" and isinstance(scenarios, list):
                    scenario_names = [str(s) for s in scenarios if s]

                # Respond immediately; the callback runs in a thread
                self._send_json_response({"status": "accepted"})

                # Run the callback in a background thread so we don't block
                thread = threading.Thread(target=cb, args=(scenario_names,), daemon=True)
                thread.start()

            def _handle_sse(self, sstate: _ServerState) -> None:
                """Handle an SSE connection — streams events to the client."""
                self.send_response(200)
                self.send_header("Content-Type", "text/event-stream")
                self.send_header("Cache-Control", "no-cache")
                self.send_header("Connection", "keep-alive")
                self.end_headers()

                client_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=64)
                with sstate.clients_lock:
                    sstate.sse_clients.append(client_queue)

                # Send initial state
                initial = {
                    "type": "state",
                    "running": sstate.running,
                    "watching": sstate.watching,
                    "canRun": sstate.can_run,
                    "autoRun": sstate.auto_run,
                    "progress": sstate.progress,
                    "progressStart": sstate.progress_start,
                }
                try:
                    self._sse_send(initial)
                except ConnectionError:
                    with sstate.clients_lock:
                        if client_queue in sstate.sse_clients:
                            sstate.sse_clients.remove(client_queue)
                    return

                try:
                    while True:
                        try:
                            event = client_queue.get(timeout=_SSE_HEARTBEAT_S)
                            self._sse_send(event)
                        except queue.Empty:
                            # Send heartbeat to keep connection alive
                            self.wfile.write(b": heartbeat\n\n")
                            self.wfile.flush()
                except ConnectionError:
                    pass
                finally:
                    with sstate.clients_lock:
                        if client_queue in sstate.sse_clients:
                            sstate.sse_clients.remove(client_queue)

            def _sse_send(self, event: dict[str, Any]) -> None:
                """Send a single SSE event."""
                data = f"data: {json.dumps(event)}\n\n"
                self.wfile.write(data.encode())
                self.wfile.flush()

            def _serve_path(self, relative: str) -> None:
                """Serve a file from assets dir with path-traversal protection."""
                target = (_ASSETS_DIR / relative).resolve()
                try:
                    target.relative_to(_ASSETS_DIR.resolve())
                except ValueError:
                    self.send_error(404)
                    return
                self._serve_file(target)

            def _serve_source(self, query: str, base: Path) -> None:
                """Serve a source code snippet around a given line.

                Query params:
                    path:    relative file path (e.g. "steps/calculator.py")
                    line:    line number to center the snippet on
                    context: number of context lines before/after (default 5)
                """
                params = parse_qs(query)
                file_path = params.get("path", [""])[0]
                line_str = params.get("line", ["0"])[0]
                context_str = params.get("context", [str(_DEFAULT_CONTEXT_LINES)])[0]

                if not file_path:
                    self._send_json_response({"error": "Missing 'path' parameter"}, status=400)
                    return

                try:
                    line = int(line_str)
                    context = max(0, int(context_str))
                except ValueError:
                    self._send_json_response(
                        {"error": "Invalid 'line' or 'context' parameter"}, status=400
                    )
                    return

                # Resolve path relative to base_dir with traversal protection
                candidate = (base / file_path).resolve()
                try:
                    candidate.relative_to(base)
                except ValueError:
                    self._send_json_response({"error": "Path outside base directory"}, status=403)
                    return

                if not candidate.exists() or not candidate.is_file():
                    self._send_json_response({"error": f"File not found: {file_path}"}, status=404)
                    return

                try:
                    lines = candidate.read_text(encoding="utf-8", errors="replace").splitlines()
                except OSError as exc:
                    self._send_json_response({"error": f"Cannot read file: {exc}"}, status=500)
                    return

                total_lines = len(lines)
                # Clamp line to valid range (1-indexed in the location)
                target_line = max(1, min(line, total_lines))
                start = max(0, target_line - 1 - context)
                end = min(total_lines, target_line + context)

                snippet_lines: list[dict[str, Any]] = []
                for i in range(start, end):
                    snippet_lines.append(
                        {
                            "number": i + 1,
                            "content": lines[i],
                            "highlight": (i + 1) == target_line,
                        }
                    )

                language = _SOURCE_EXTENSIONS.get(candidate.suffix, "text")

                payload = {
                    "path": file_path,
                    "line": target_line,
                    "language": language,
                    "snippet": snippet_lines,
                    "total_lines": total_lines,
                }
                self._send_json_response(payload)

            def _send_json_response(self, data: dict[str, Any], status: int = 200) -> None:
                """Send a JSON response with optional gzip."""
                body = json.dumps(data).encode()
                accept_gzip = "gzip" in (self.headers.get("Accept-Encoding", ""))
                if accept_gzip and len(body) > _GZIP_THRESHOLD:
                    body = gzip.compress(body)
                    self.send_response(status)
                    self.send_header("Content-Type", "application/json; charset=utf-8")
                    self.send_header("Content-Encoding", "gzip")
                    self.send_header("Content-Length", str(len(body)))
                    self.end_headers()
                    self.wfile.write(body)
                    return
                self.send_response(status)
                self.send_header("Content-Type", "application/json; charset=utf-8")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def _send_json(self, raw: bytes, gzipped: bytes) -> None:
                accept_gzip = "gzip" in (self.headers.get("Accept-Encoding", ""))
                body = gzipped if accept_gzip else raw
                self.send_response(200)
                self.send_header("Content-Type", "application/json; charset=utf-8")
                self.send_header("Cache-Control", "no-cache")
                self.send_header("Content-Length", str(len(body)))
                if accept_gzip:
                    self.send_header("Content-Encoding", "gzip")
                self.end_headers()
                self.wfile.write(body)

            def _serve_file(self, path: Path) -> None:
                if not path.exists() or not path.is_file():
                    self.send_error(404)
                    return
                mime = _MIME_TYPES.get(path.suffix, "application/octet-stream")
                data = path.read_bytes()
                if mime in _COMPRESSIBLE_TYPES:
                    accept_gzip = "gzip" in (self.headers.get("Accept-Encoding", ""))
                    if accept_gzip and len(data) > _GZIP_THRESHOLD:
                        data = gzip.compress(data)
                        self.send_response(200)
                        self.send_header("Content-Type", mime)
                        self.send_header("Content-Encoding", "gzip")
                        self.send_header("Content-Length", str(len(data)))
                        self.end_headers()
                        self.wfile.write(data)
                        return
                self.send_response(200)
                self.send_header("Content-Type", mime)
                self.send_header("Content-Length", str(len(data)))
                self.end_headers()
                self.wfile.write(data)

            def log_message(self, *args: Any) -> None:
                pass

        self._httpd = ThreadingHTTPServer(("127.0.0.1", self.port), Handler)
        actual_port = self._httpd.server_address[1]
        self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True)
        self._thread.start()
        return f"http://127.0.0.1:{actual_port}"

    def wait(self) -> None:
        """Block until the server stops."""
        if self._thread is not None:
            self._thread.join()

    def stop(self) -> None:
        """Stop the server."""
        if self._httpd is not None:
            self._httpd.shutdown()
            self._httpd.server_close()
            self._httpd = None

    def update_trace(self, trace: Trace) -> None:
        """Update the trace served by this server and notify SSE clients."""
        self.trace = trace
        trace_json = json.dumps(as_dict(trace), default=str).encode()
        self._state.update_trace(trace_json)

    def set_running(self, running: bool) -> None:
        """Set the running state and notify SSE clients."""
        self._state.set_running(running)

    def set_auto_run(self, auto_run: bool) -> None:
        """Set the auto-run state and notify SSE clients."""
        self._state.auto_run = auto_run
        self._state.notify({"type": "state", "autoRun": auto_run})

    def get_auto_run(self) -> bool:
        """Return the current auto-run state."""
        return self._state.auto_run

    def notify(self, event: dict[str, Any]) -> None:
        """Push a custom event to all SSE clients."""
        self._state.notify(event)

url property 

url: str

Return the URL the server is listening on.

get_auto_run 

get_auto_run() -> bool

Return the current auto-run state.

Source code in behave_trace/viewer/server.py 
551
552
553
def get_auto_run(self) -> bool:
    """Return the current auto-run state."""
    return self._state.auto_run

notify 

notify(event: dict[str, Any]) -> None

Push a custom event to all SSE clients.

Source code in behave_trace/viewer/server.py 
555
556
557
def notify(self, event: dict[str, Any]) -> None:
    """Push a custom event to all SSE clients."""
    self._state.notify(event)

set_auto_run 

set_auto_run(auto_run: bool) -> None

Set the auto-run state and notify SSE clients.

Source code in behave_trace/viewer/server.py 
546
547
548
549
def set_auto_run(self, auto_run: bool) -> None:
    """Set the auto-run state and notify SSE clients."""
    self._state.auto_run = auto_run
    self._state.notify({"type": "state", "autoRun": auto_run})

set_running 

set_running(running: bool) -> None

Set the running state and notify SSE clients.

Source code in behave_trace/viewer/server.py 
542
543
544
def set_running(self, running: bool) -> None:
    """Set the running state and notify SSE clients."""
    self._state.set_running(running)

start 

start() -> str

Start the server and return the URL.

Source code in behave_trace/viewer/server.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
def start(self) -> str:
    """Start the server and return the URL."""
    state = self._state
    base_dir = self._base_dir
    rerun_cb = self.rerun_callback

    class Handler(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def do_GET(self) -> None:
            parsed = urlparse(self.path)
            if parsed.path == "/api/trace":
                self._send_json(state.trace_bytes, state.trace_gzipped)
            elif parsed.path == "/api/source":
                self._serve_source(parsed.query, base_dir)
            elif parsed.path == "/api/watching":
                payload = json.dumps({"watching": state.watching}).encode()
                self._send_json(payload, gzip.compress(payload))
            elif parsed.path == "/api/stream":
                self._handle_sse(state)
            elif parsed.path == "/":
                self._serve_file(_ASSETS_DIR / "index.html")
            elif parsed.path.startswith("/css/") or parsed.path.startswith("/js/"):
                self._serve_path(parsed.path.lstrip("/"))
            else:
                self.send_error(404)

        def do_POST(self) -> None:
            parsed = urlparse(self.path)
            if parsed.path == "/api/rerun":
                self._handle_rerun(state, rerun_cb)
            elif parsed.path == "/api/run":
                self._handle_run(state, rerun_cb)
            elif parsed.path == "/api/autorun":
                self._handle_autorun(state)
            elif parsed.path == "/api/progress":
                self._handle_progress(state)
            else:
                self._send_json_response({"error": "Not found"}, status=404)

        def _handle_progress(self, sstate: _ServerState) -> None:
            """Handle POST /api/progress — update live progress."""
            try:
                content_length = int(self.headers.get("Content-Length", 0))
            except (ValueError, TypeError):
                content_length = 0
            body = self.rfile.read(content_length) if content_length > 0 else b"{}"
            try:
                payload = json.loads(body)
            except json.JSONDecodeError:
                self._send_json_response({"error": "Invalid JSON"}, status=400)
                return

            if not isinstance(payload, dict):
                self._send_json_response({"error": "Expected JSON object"}, status=400)
                return

            completed = payload.get("completed", sstate.progress["completed"])
            total = payload.get("total", sstate.progress["total"])
            sstate.progress = {
                "completed": int(completed),
                "total": int(total),
            }
            sstate.notify(
                {
                    "type": "scenario_completed",
                    "completed": sstate.progress["completed"],
                    "total": sstate.progress["total"],
                    "scenario_name": payload.get("scenario_name", ""),
                }
            )
            self._send_json_response({"status": "ok"})

        def _handle_autorun(self, sstate: _ServerState) -> None:
            """Handle POST /api/autorun — toggle auto-run on/off."""
            try:
                content_length = int(self.headers.get("Content-Length", 0))
            except (ValueError, TypeError):
                content_length = 0
            body = self.rfile.read(content_length) if content_length > 0 else b"{}"
            try:
                payload = json.loads(body)
            except json.JSONDecodeError:
                self._send_json_response({"error": "Invalid JSON"}, status=400)
                return

            if not isinstance(payload, dict):
                self._send_json_response({"error": "Expected JSON object"}, status=400)
                return

            enabled = payload.get("enabled")
            if not isinstance(enabled, bool):
                self._send_json_response({"error": "Missing 'enabled' boolean"}, status=400)
                return

            sstate.auto_run = enabled
            sstate.notify({"type": "state", "autoRun": enabled})
            self._send_json_response({"status": "ok", "autoRun": enabled})

        def _handle_run(
            self,
            sstate: _ServerState,
            cb: Callable[[list[str] | None], None] | None,
        ) -> None:
            """Handle POST /api/run — execute behave from scratch (all tests)."""
            if cb is None:
                self._send_json_response(
                    {"error": "Run not available (no callback configured)"},
                    status=501,
                )
                return
            if sstate.running:
                self._send_json_response({"error": "Already running"}, status=409)
                return
            self._send_json_response({"status": "accepted"})
            thread = threading.Thread(target=cb, args=(None,), daemon=True)
            thread.start()

        def _handle_rerun(
            self,
            sstate: _ServerState,
            cb: Callable[[list[str] | None], None] | None,
        ) -> None:
            """Handle POST /api/rerun — re-execute behave with optional filter."""
            if cb is None:
                self._send_json_response(
                    {"error": "Re-run not available (no callback configured)"},
                    status=501,
                )
                return

            try:
                content_length = int(self.headers.get("Content-Length", 0))
            except (ValueError, TypeError):
                content_length = 0
            body = self.rfile.read(content_length) if content_length > 0 else b"{}"
            try:
                payload = json.loads(body)
            except json.JSONDecodeError:
                self._send_json_response({"error": "Invalid JSON"}, status=400)
                return

            if not isinstance(payload, dict):
                self._send_json_response({"error": "Expected JSON object"}, status=400)
                return

            filter_type = payload.get("filter", "all")
            scenarios = payload.get("scenarios")

            if filter_type not in ("failed", "all"):
                self._send_json_response(
                    {"error": "Invalid filter; must be 'failed' or 'all'"},
                    status=400,
                )
                return

            scenario_names: list[str] | None = None
            if filter_type == "failed" and isinstance(scenarios, list):
                scenario_names = [str(s) for s in scenarios if s]

            # Respond immediately; the callback runs in a thread
            self._send_json_response({"status": "accepted"})

            # Run the callback in a background thread so we don't block
            thread = threading.Thread(target=cb, args=(scenario_names,), daemon=True)
            thread.start()

        def _handle_sse(self, sstate: _ServerState) -> None:
            """Handle an SSE connection — streams events to the client."""
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Connection", "keep-alive")
            self.end_headers()

            client_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=64)
            with sstate.clients_lock:
                sstate.sse_clients.append(client_queue)

            # Send initial state
            initial = {
                "type": "state",
                "running": sstate.running,
                "watching": sstate.watching,
                "canRun": sstate.can_run,
                "autoRun": sstate.auto_run,
                "progress": sstate.progress,
                "progressStart": sstate.progress_start,
            }
            try:
                self._sse_send(initial)
            except ConnectionError:
                with sstate.clients_lock:
                    if client_queue in sstate.sse_clients:
                        sstate.sse_clients.remove(client_queue)
                return

            try:
                while True:
                    try:
                        event = client_queue.get(timeout=_SSE_HEARTBEAT_S)
                        self._sse_send(event)
                    except queue.Empty:
                        # Send heartbeat to keep connection alive
                        self.wfile.write(b": heartbeat\n\n")
                        self.wfile.flush()
            except ConnectionError:
                pass
            finally:
                with sstate.clients_lock:
                    if client_queue in sstate.sse_clients:
                        sstate.sse_clients.remove(client_queue)

        def _sse_send(self, event: dict[str, Any]) -> None:
            """Send a single SSE event."""
            data = f"data: {json.dumps(event)}\n\n"
            self.wfile.write(data.encode())
            self.wfile.flush()

        def _serve_path(self, relative: str) -> None:
            """Serve a file from assets dir with path-traversal protection."""
            target = (_ASSETS_DIR / relative).resolve()
            try:
                target.relative_to(_ASSETS_DIR.resolve())
            except ValueError:
                self.send_error(404)
                return
            self._serve_file(target)

        def _serve_source(self, query: str, base: Path) -> None:
            """Serve a source code snippet around a given line.

            Query params:
                path:    relative file path (e.g. "steps/calculator.py")
                line:    line number to center the snippet on
                context: number of context lines before/after (default 5)
            """
            params = parse_qs(query)
            file_path = params.get("path", [""])[0]
            line_str = params.get("line", ["0"])[0]
            context_str = params.get("context", [str(_DEFAULT_CONTEXT_LINES)])[0]

            if not file_path:
                self._send_json_response({"error": "Missing 'path' parameter"}, status=400)
                return

            try:
                line = int(line_str)
                context = max(0, int(context_str))
            except ValueError:
                self._send_json_response(
                    {"error": "Invalid 'line' or 'context' parameter"}, status=400
                )
                return

            # Resolve path relative to base_dir with traversal protection
            candidate = (base / file_path).resolve()
            try:
                candidate.relative_to(base)
            except ValueError:
                self._send_json_response({"error": "Path outside base directory"}, status=403)
                return

            if not candidate.exists() or not candidate.is_file():
                self._send_json_response({"error": f"File not found: {file_path}"}, status=404)
                return

            try:
                lines = candidate.read_text(encoding="utf-8", errors="replace").splitlines()
            except OSError as exc:
                self._send_json_response({"error": f"Cannot read file: {exc}"}, status=500)
                return

            total_lines = len(lines)
            # Clamp line to valid range (1-indexed in the location)
            target_line = max(1, min(line, total_lines))
            start = max(0, target_line - 1 - context)
            end = min(total_lines, target_line + context)

            snippet_lines: list[dict[str, Any]] = []
            for i in range(start, end):
                snippet_lines.append(
                    {
                        "number": i + 1,
                        "content": lines[i],
                        "highlight": (i + 1) == target_line,
                    }
                )

            language = _SOURCE_EXTENSIONS.get(candidate.suffix, "text")

            payload = {
                "path": file_path,
                "line": target_line,
                "language": language,
                "snippet": snippet_lines,
                "total_lines": total_lines,
            }
            self._send_json_response(payload)

        def _send_json_response(self, data: dict[str, Any], status: int = 200) -> None:
            """Send a JSON response with optional gzip."""
            body = json.dumps(data).encode()
            accept_gzip = "gzip" in (self.headers.get("Accept-Encoding", ""))
            if accept_gzip and len(body) > _GZIP_THRESHOLD:
                body = gzip.compress(body)
                self.send_response(status)
                self.send_header("Content-Type", "application/json; charset=utf-8")
                self.send_header("Content-Encoding", "gzip")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)
                return
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def _send_json(self, raw: bytes, gzipped: bytes) -> None:
            accept_gzip = "gzip" in (self.headers.get("Accept-Encoding", ""))
            body = gzipped if accept_gzip else raw
            self.send_response(200)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Content-Length", str(len(body)))
            if accept_gzip:
                self.send_header("Content-Encoding", "gzip")
            self.end_headers()
            self.wfile.write(body)

        def _serve_file(self, path: Path) -> None:
            if not path.exists() or not path.is_file():
                self.send_error(404)
                return
            mime = _MIME_TYPES.get(path.suffix, "application/octet-stream")
            data = path.read_bytes()
            if mime in _COMPRESSIBLE_TYPES:
                accept_gzip = "gzip" in (self.headers.get("Accept-Encoding", ""))
                if accept_gzip and len(data) > _GZIP_THRESHOLD:
                    data = gzip.compress(data)
                    self.send_response(200)
                    self.send_header("Content-Type", mime)
                    self.send_header("Content-Encoding", "gzip")
                    self.send_header("Content-Length", str(len(data)))
                    self.end_headers()
                    self.wfile.write(data)
                    return
            self.send_response(200)
            self.send_header("Content-Type", mime)
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        def log_message(self, *args: Any) -> None:
            pass

    self._httpd = ThreadingHTTPServer(("127.0.0.1", self.port), Handler)
    actual_port = self._httpd.server_address[1]
    self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True)
    self._thread.start()
    return f"http://127.0.0.1:{actual_port}"

stop 

stop() -> None

Stop the server.

Source code in behave_trace/viewer/server.py 
529
530
531
532
533
534
def stop(self) -> None:
    """Stop the server."""
    if self._httpd is not None:
        self._httpd.shutdown()
        self._httpd.server_close()
        self._httpd = None

update_trace 

update_trace(trace: Trace) -> None

Update the trace served by this server and notify SSE clients.

Source code in behave_trace/viewer/server.py 
536
537
538
539
540
def update_trace(self, trace: Trace) -> None:
    """Update the trace served by this server and notify SSE clients."""
    self.trace = trace
    trace_json = json.dumps(as_dict(trace), default=str).encode()
    self._state.update_trace(trace_json)

wait 

wait() -> None

Block until the server stops.

Source code in behave_trace/viewer/server.py 
524
525
526
527
def wait(self) -> None:
    """Block until the server stops."""
    if self._thread is not None:
        self._thread.join()

behave_trace.viewer.browser.open_app 

open_app(url: str) -> None

Open the given URL in an app-like browser window.

Tries chrome --app first, falls back to webbrowser.open().

Source code in behave_trace/viewer/browser.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def open_app(url: str) -> None:
    """Open the given URL in an app-like browser window.

    Tries chrome --app first, falls back to webbrowser.open().
    """
    chrome = _find_chrome()
    if chrome:
        try:
            subprocess.Popen(
                [chrome, "--app=" + url, "--new-window"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
            return
        except Exception:
            pass
    webbrowser.open(url)