Coverage for pytest_beehave/sync_engine.py: 100%
158 statements
« prev ^ index » next coverage.py v7.8.0, created at 2026-04-21 04:49 +0000
1"""Sync engine for pytest-beehave — orchestrates stub creation and updates."""
3from __future__ import annotations
5from dataclasses import dataclass
6from pathlib import Path
7from typing import Protocol
9from pytest_beehave.config import StubFormat
10from pytest_beehave.feature_parser import (
11 ParsedExample,
12 ParsedFeature,
13 ParsedRule,
14 parse_feature,
15)
16from pytest_beehave.models import ExampleId, FeatureStage
17from pytest_beehave.stub_reader import ExistingStub, read_stubs_from_file
18from pytest_beehave.stub_writer import (
19 StubSpec,
20 SyncAction,
21 build_class_name,
22 build_docstring,
23 build_function_name,
24 mark_non_conforming,
25 mark_orphan,
26 remove_orphan_marker,
27 toggle_deprecated_marker,
28 update_docstring,
29 write_stub_to_file,
30)
32_FEATURE_STAGES = (
33 FeatureStage.BACKLOG,
34 FeatureStage.IN_PROGRESS,
35 FeatureStage.COMPLETED,
36)
39class FileSystemProtocol(Protocol):
40 """Protocol for filesystem operations needed by the sync engine."""
42 def list_feature_files(self, stage_dir: Path) -> list[Path]: # pragma: no cover
43 """List all .feature files recursively under stage_dir."""
44 ...
46 def list_test_files(self, tests_dir: Path) -> list[Path]: # pragma: no cover
47 """List all *_test.py files recursively under tests_dir."""
48 ...
51@dataclass(frozen=True, slots=True)
52class _RealFileSystem:
53 """Concrete filesystem adapter using pathlib."""
55 def list_feature_files(self, stage_dir: Path) -> list[Path]:
56 """List all .feature files recursively under stage_dir.
58 Args:
59 stage_dir: Root directory to search.
61 Returns:
62 Sorted list of .feature file paths.
63 """
64 return sorted(stage_dir.rglob("*.feature"))
66 def list_test_files(self, tests_dir: Path) -> list[Path]:
67 """List all *_test.py files recursively under tests_dir.
69 Args:
70 tests_dir: Root directory to search.
72 Returns:
73 Sorted list of test file paths.
74 """
75 return sorted(tests_dir.rglob("*_test.py"))
78@dataclass(frozen=True, slots=True)
79class SyncResult:
80 """Result of a sync operation.
82 Attributes:
83 actions: Tuple of SyncAction objects describing what was done.
84 """
86 actions: tuple[SyncAction, ...]
88 @property
89 def is_noop(self) -> bool:
90 """Return True if no actions were taken."""
91 return len(self.actions) == 0
94def _collect_all_ids(
95 features_dir: Path,
96 filesystem: FileSystemProtocol,
97) -> frozenset[ExampleId]:
98 """Collect all example IDs from all .feature files.
100 Args:
101 features_dir: Root of the features directory.
102 filesystem: Filesystem adapter.
104 Returns:
105 Frozenset of ExampleId objects.
106 """
107 import re
109 id_re = re.compile(r"@id:([a-f0-9]{8})")
110 ids: set[ExampleId] = set()
111 for stage in _FEATURE_STAGES:
112 stage_dir = features_dir / stage.value
113 if not stage_dir.exists():
114 continue
115 for path in filesystem.list_feature_files(stage_dir):
116 text = path.read_text(encoding="utf-8")
117 ids.update(ExampleId(match.group(1)) for match in id_re.finditer(text))
118 return frozenset(ids)
121def _add_rule_locations(
122 feature_dir: Path,
123 feature: ParsedFeature,
124 locations: dict[ExampleId, tuple[Path, str | None]],
125) -> None:
126 """Add expected locations for all rule-based examples of a feature.
128 Args:
129 feature_dir: The test directory for this feature.
130 feature: The parsed feature.
131 locations: Dict to populate with ExampleId -> (file, class) entries.
132 """
133 for rule in feature.rules:
134 test_file = feature_dir / f"{rule.rule_slug}_test.py"
135 class_name = build_class_name(rule.rule_slug)
136 for example in rule.examples:
137 locations[example.example_id] = (test_file, class_name)
140def _build_expected_locations(
141 feature_stage_pairs: list[tuple[ParsedFeature, FeatureStage]],
142 tests_dir: Path,
143) -> dict[ExampleId, tuple[Path, str | None]]:
144 """Build a map of ExampleId to (expected_test_file, expected_class_name).
146 Args:
147 feature_stage_pairs: All parsed feature/stage tuples.
148 tests_dir: Root of the tests/features/ directory.
150 Returns:
151 Dict mapping ExampleId to (file_path, class_name_or_None).
152 """
153 locations: dict[ExampleId, tuple[Path, str | None]] = {}
154 for feature, _stage in feature_stage_pairs:
155 feature_dir = tests_dir / str(feature.feature_slug)
156 _add_rule_locations(feature_dir, feature, locations)
157 top_level_file = feature_dir / "examples_test.py"
158 for example in feature.top_level_examples:
159 locations[example.example_id] = (top_level_file, None)
160 return locations
163def _orphan_action(
164 test_file: Path,
165 stub: ExistingStub,
166 all_ids: frozenset[ExampleId],
167) -> SyncAction | None:
168 """Compute the orphan sync action for a single stub.
170 Args:
171 test_file: The test file containing the stub.
172 stub: The existing stub.
173 all_ids: All known example IDs.
175 Returns:
176 SyncAction or None.
177 """
178 if stub.example_id not in all_ids:
179 return mark_orphan(test_file, stub.function_name)
180 return remove_orphan_marker(test_file, stub.function_name)
183def _orphan_actions_for_file(
184 test_file: Path,
185 all_ids: frozenset[ExampleId],
186) -> list[SyncAction]:
187 """Compute orphan sync actions for all stubs in a single test file.
189 Args:
190 test_file: The test file to scan.
191 all_ids: All known example IDs.
193 Returns:
194 List of SyncAction objects.
195 """
196 return [
197 action
198 for stub in read_stubs_from_file(test_file)
199 for action in [_orphan_action(test_file, stub, all_ids)]
200 if action is not None
201 ]
204def _sync_orphans(
205 tests_dir: Path,
206 all_ids: frozenset[ExampleId],
207 filesystem: FileSystemProtocol,
208) -> list[SyncAction]:
209 """Add or remove orphan markers for test functions without matching @id.
211 Args:
212 tests_dir: Root of the tests/features/ directory.
213 all_ids: All known example IDs.
214 filesystem: Filesystem adapter.
216 Returns:
217 List of SyncAction objects.
218 """
219 actions: list[SyncAction] = []
220 for test_file in filesystem.list_test_files(tests_dir):
221 actions.extend(_orphan_actions_for_file(test_file, all_ids))
222 return actions
225def _check_stub_conformity(
226 test_file: Path,
227 stub: ExistingStub,
228 expected_locations: dict[ExampleId, tuple[Path, str | None]],
229) -> SyncAction | None:
230 """Check if a single stub is in the correct file and mark it if not.
232 Args:
233 test_file: The file the stub was found in.
234 stub: The existing stub.
235 expected_locations: Map of ExampleId to (expected_file, expected_class).
237 Returns:
238 SyncAction if non-conforming, else None.
239 """
240 if stub.example_id not in expected_locations:
241 return None
242 expected_file, expected_class = expected_locations[stub.example_id]
243 if expected_class is None or test_file == expected_file:
244 return None
245 return mark_non_conforming(
246 test_file, stub.function_name, expected_file, expected_class
247 )
250def _non_conforming_actions_for_file(
251 test_file: Path,
252 expected_locations: dict[ExampleId, tuple[Path, str | None]],
253) -> list[SyncAction]:
254 """Collect non-conforming actions for all stubs in one test file.
256 Args:
257 test_file: Path to the test file.
258 expected_locations: Map of ExampleId to (expected_file, expected_class).
260 Returns:
261 List of SyncAction objects for non-conforming stubs.
262 """
263 return [
264 action
265 for stub in read_stubs_from_file(test_file)
266 for action in [_check_stub_conformity(test_file, stub, expected_locations)]
267 if action is not None
268 ]
271def _sync_non_conforming(
272 tests_dir: Path,
273 expected_locations: dict[ExampleId, tuple[Path, str | None]],
274 filesystem: FileSystemProtocol,
275) -> list[SyncAction]:
276 """Mark test stubs found in the wrong file as non-conforming.
278 Only applies to Rule-based stubs (those with an expected class). Top-level
279 examples without a class context are skipped — orphan detection handles
280 unrecognised test files.
282 Args:
283 tests_dir: Root of the tests/features/ directory.
284 expected_locations: Map of ExampleId to (expected_file, expected_class).
285 filesystem: Filesystem adapter.
287 Returns:
288 List of SyncAction objects.
289 """
290 actions: list[SyncAction] = []
291 for test_file in filesystem.list_test_files(tests_dir):
292 actions.extend(_non_conforming_actions_for_file(test_file, expected_locations))
293 return actions
296def _sync_active_feature(
297 feature: ParsedFeature,
298 tests_dir: Path,
299 stub_format: StubFormat = "functions",
300) -> list[SyncAction]:
301 """Sync stubs for an active (backlog/in-progress) feature.
303 Args:
304 feature: The parsed feature.
305 tests_dir: Root of the tests/features/ directory.
306 stub_format: The output format for new stubs.
308 Returns:
309 List of SyncAction objects.
310 """
311 actions: list[SyncAction] = []
312 feature_test_dir = tests_dir / str(feature.feature_slug)
314 if feature.rules:
315 for rule in feature.rules:
316 actions.extend(
317 _sync_rule_stubs(feature, rule, feature_test_dir, stub_format)
318 )
319 elif feature.top_level_examples:
320 actions.extend(_sync_top_level_stubs(feature, feature_test_dir))
322 return actions
325def _sync_rule_stubs(
326 feature: ParsedFeature,
327 rule: ParsedRule,
328 feature_test_dir: Path,
329 stub_format: StubFormat = "functions",
330) -> list[SyncAction]:
331 """Sync stubs for a single rule block.
333 Args:
334 feature: The parsed feature.
335 rule: The parsed rule.
336 feature_test_dir: Directory for this feature's tests.
337 stub_format: The output format for new stubs.
339 Returns:
340 List of SyncAction objects.
341 """
342 if not rule.examples:
343 return []
344 test_file = feature_test_dir / f"{rule.rule_slug}_test.py"
345 existing = {s.example_id: s for s in read_stubs_from_file(test_file)}
346 actions: list[SyncAction] = []
347 for example in rule.examples:
348 action = _sync_one_example(
349 feature, rule, example, test_file, existing, stub_format
350 )
351 if action is not None:
352 actions.append(action)
353 actions.extend(_sync_deprecated_in_rule(feature, rule, test_file))
354 return actions
357def _update_existing_stub(
358 feature: ParsedFeature,
359 rule: ParsedRule | None,
360 example: ParsedExample,
361 test_file: Path,
362 stub: ExistingStub,
363) -> SyncAction | None:
364 """Update an existing stub's docstring and/or name.
366 Args:
367 feature: The parsed feature.
368 rule: The parsed rule (or None for top-level).
369 example: The parsed example.
370 test_file: Path to the test file.
371 stub: The existing stub to update.
373 Returns:
374 SyncAction or None.
375 """
376 return update_docstring(
377 test_file,
378 stub.function_name,
379 build_docstring(feature, rule, example),
380 feature.feature_slug,
381 example.example_id,
382 )
385def _sync_one_example(
386 feature: ParsedFeature,
387 rule: ParsedRule | None,
388 example: ParsedExample,
389 test_file: Path,
390 existing: dict[ExampleId, ExistingStub],
391 stub_format: StubFormat = "functions",
392) -> SyncAction | None:
393 """Sync a single example stub — create or update.
395 Args:
396 feature: The parsed feature.
397 rule: The parsed rule (or None for top-level).
398 example: The parsed example.
399 test_file: Path to the test file.
400 existing: Map of existing stubs by example ID.
401 stub_format: The output format for new stubs.
403 Returns:
404 SyncAction or None.
405 """
406 if example.example_id in existing:
407 return _update_existing_stub(
408 feature, rule, example, test_file, existing[example.example_id]
409 )
410 rule_slug = rule.rule_slug if rule else None
411 spec = StubSpec(
412 feature_slug=feature.feature_slug,
413 rule_slug=rule_slug,
414 example=example,
415 feature=feature,
416 stub_format=stub_format,
417 )
418 return write_stub_to_file(test_file, spec)
421def _sync_deprecated_in_rule(
422 feature: ParsedFeature,
423 rule: ParsedRule,
424 test_file: Path,
425) -> list[SyncAction]:
426 """Sync deprecated markers for all examples in a rule.
428 Args:
429 feature: The parsed feature.
430 rule: The parsed rule.
431 test_file: Path to the test file.
433 Returns:
434 List of SyncAction objects.
435 """
436 return [
437 action
438 for example in rule.examples
439 for action in [
440 toggle_deprecated_marker(
441 test_file,
442 build_function_name(feature.feature_slug, example.example_id),
443 should_be_deprecated=example.is_deprecated,
444 )
445 ]
446 if action is not None
447 ]
450def _sync_top_level_stubs(
451 feature: ParsedFeature,
452 feature_test_dir: Path,
453) -> list[SyncAction]:
454 """Sync stubs for top-level examples (no Rule blocks).
456 Args:
457 feature: The parsed feature.
458 feature_test_dir: Directory for this feature's tests.
460 Returns:
461 List of SyncAction objects.
462 """
463 test_file = feature_test_dir / "examples_test.py"
464 existing = {s.example_id: s for s in read_stubs_from_file(test_file)}
465 actions: list[SyncAction] = []
466 for example in feature.top_level_examples:
467 action = _sync_one_example(feature, None, example, test_file, existing)
468 if action is not None:
469 actions.append(action)
470 actions.extend(_sync_deprecated_top_level(feature, test_file))
471 return actions
474def _sync_deprecated_top_level(
475 feature: ParsedFeature,
476 test_file: Path,
477) -> list[SyncAction]:
478 """Sync deprecated markers for top-level examples (no Rule blocks).
480 Args:
481 feature: The parsed feature.
482 test_file: Path to the top-level examples test file.
484 Returns:
485 List of SyncAction objects.
486 """
487 return [
488 action
489 for example in feature.top_level_examples
490 for action in [
491 toggle_deprecated_marker(
492 test_file,
493 build_function_name(feature.feature_slug, example.example_id),
494 should_be_deprecated=example.is_deprecated,
495 )
496 ]
497 if action is not None
498 ]
501def _sync_deprecated_rules(
502 feature: ParsedFeature,
503 feature_test_dir: Path,
504) -> list[SyncAction]:
505 """Sync deprecated markers for all rules in a feature.
507 Args:
508 feature: The parsed feature.
509 feature_test_dir: Directory for this feature's tests.
511 Returns:
512 List of SyncAction objects.
513 """
514 actions: list[SyncAction] = []
515 for rule in feature.rules:
516 test_file = feature_test_dir / f"{rule.rule_slug}_test.py"
517 actions.extend(_sync_deprecated_in_rule(feature, rule, test_file))
518 return actions
521def _sync_completed_feature(
522 feature: ParsedFeature,
523 tests_dir: Path,
524) -> list[SyncAction]:
525 """Sync deprecated markers only for a completed feature.
527 Args:
528 feature: The parsed feature.
529 tests_dir: Root of the tests/features/ directory.
531 Returns:
532 List of SyncAction objects.
533 """
534 feature_test_dir = tests_dir / str(feature.feature_slug)
535 if feature.rules:
536 return _sync_deprecated_rules(feature, feature_test_dir)
537 return _sync_deprecated_top_level(feature, feature_test_dir / "examples_test.py")
540def _folder_name_for(feature_path: Path, stage_dir: Path) -> str:
541 """Derive the folder name for a feature file.
543 Args:
544 feature_path: Path to the .feature file.
545 stage_dir: The stage directory (backlog/, in-progress/, completed/).
547 Returns:
548 Folder name string used as the feature slug source.
549 """
550 feature_parent = feature_path.parent
551 if feature_parent == stage_dir:
552 return feature_path.stem
553 return feature_parent.name
556def _features_in_stage(
557 stage: FeatureStage,
558 features_dir: Path,
559 filesystem: FileSystemProtocol,
560) -> list[tuple[ParsedFeature, FeatureStage]]:
561 """Discover all feature files for one stage directory.
563 Args:
564 stage: The feature stage.
565 features_dir: Root of the features directory.
566 filesystem: Filesystem adapter.
568 Returns:
569 List of (ParsedFeature, FeatureStage) tuples for this stage.
570 """
571 stage_dir = features_dir / stage.value
572 if not stage_dir.exists():
573 return []
574 return [
575 (parse_feature(p, _folder_name_for(p, stage_dir)), stage)
576 for p in filesystem.list_feature_files(stage_dir)
577 ]
580def discover_feature_locations(
581 features_dir: Path,
582 filesystem: FileSystemProtocol,
583) -> list[tuple[ParsedFeature, FeatureStage]]:
584 """Discover all feature files and their stages.
586 Args:
587 features_dir: Root of the features directory.
588 filesystem: Filesystem adapter.
590 Returns:
591 List of (ParsedFeature, FeatureStage) tuples.
592 """
593 results: list[tuple[ParsedFeature, FeatureStage]] = []
594 for stage in _FEATURE_STAGES:
595 results.extend(_features_in_stage(stage, features_dir, filesystem))
596 return results
599def run_sync(
600 features_root: Path,
601 tests_root: Path,
602 filesystem: FileSystemProtocol | None = None,
603 stub_format: StubFormat = "functions",
604) -> list[str]:
605 """Sync test stubs from .feature files to the tests directory.
607 Args:
608 features_root: Root of the features directory (contains backlog/,
609 in-progress/, completed/).
610 tests_root: Root of the tests/features/ directory.
611 filesystem: Optional filesystem adapter. Defaults to _RealFileSystem.
612 stub_format: The output format for new stubs.
614 Returns:
615 List of action description strings.
616 """
617 if filesystem is None:
618 filesystem = _RealFileSystem()
619 actions: list[SyncAction] = []
620 feature_stage_pairs = discover_feature_locations(features_root, filesystem)
621 for feature, stage in feature_stage_pairs:
622 if stage == FeatureStage.COMPLETED:
623 actions.extend(_sync_completed_feature(feature, tests_root))
624 else:
625 actions.extend(_sync_active_feature(feature, tests_root, stub_format))
626 expected_locations = _build_expected_locations(feature_stage_pairs, tests_root)
627 actions.extend(_sync_non_conforming(tests_root, expected_locations, filesystem))
628 all_ids = _collect_all_ids(features_root, filesystem)
629 actions.extend(_sync_orphans(tests_root, all_ids, filesystem))
630 return [str(a) for a in actions]