Selfie
Loading...
Searching...
No Matches
pytest_selfie.plugin.SnapshotFileProgress Class Reference

Public Member Functions

 __init__ (self, PytestSnapshotSystem system, TypedPath test_file)
 assert_not_terminated (self)
 test_start (self, str testname)
 test_failed (self, str testname)
 test_finish (self, str testname)
 keep (self, str test, Optional[str] suffix_or_all)
 write (self, str test, str suffix, Snapshot snapshot, CallStack call_stack, SnapshotFileLayout layout)
Optional[Snapshot] read (self, str test, str suffix)
SnapshotFile read_file (self)

Public Attributes

 system = system
 test_file = test_file
int finishes_expected = 0
int finishes_so_far = 0
bool has_failed = False
Optional[SnapshotFile] file = None
AtomicReference[ArrayMap[str, WithinTestGC]] tests
Optional[DiskWriteTracker] disk_write_tracker = DiskWriteTracker()
Optional[str] testname_in_progress = None
bool testname_in_progress_failed = False

Static Public Attributes

 TERMINATED = ArrayMap.empty().plus(" ~ / f!n1shed / ~ ", WithinTestGC())

Detailed Description

Definition at line 287 of file plugin.py.

Constructor & Destructor Documentation

◆ __init__()

pytest_selfie.plugin.SnapshotFileProgress.__init__ ( self,
PytestSnapshotSystem system,
TypedPath test_file )

Definition at line 290 of file plugin.py.

290 def __init__(self, system: PytestSnapshotSystem, test_file: TypedPath):
291 self.system = system
292 # the test file which holds the test case which we are the snapshot file for
293 self.test_file = test_file
294
295 # before the tests run, we find out how many we expect to happen
296 self.finishes_expected = 0
297 # while the tests run, we count up until they have all run, and then we can cleanup
298 self.finishes_so_far = 0
299 # have any tests failed?
300 self.has_failed = False
301
302 # lazy-loaded snapshot file
303 self.file: Optional[SnapshotFile] = None
304 self.tests: AtomicReference[ArrayMap[str, WithinTestGC]] = AtomicReference(
305 ArrayMap.empty()
306 )
307 self.disk_write_tracker: Optional[DiskWriteTracker] = DiskWriteTracker()
308 # the test name which is currently in progress, if any
309 self.testname_in_progress: Optional[str] = None
310 self.testname_in_progress_failed = False
311

Member Function Documentation

◆ assert_not_terminated()

pytest_selfie.plugin.SnapshotFileProgress.assert_not_terminated ( self)

Definition at line 312 of file plugin.py.

312 def assert_not_terminated(self):
313 if self.tests.get() == SnapshotFileProgress.TERMINATED:
314 raise RuntimeError(
315 "Cannot call methods on a terminated SnapshotFileProgress"
316 )
317

◆ keep()

pytest_selfie.plugin.SnapshotFileProgress.keep ( self,
str test,
Optional[str] suffix_or_all )

Definition at line 401 of file plugin.py.

401 def keep(self, test: str, suffix_or_all: Optional[str]):
402 self.assert_not_terminated()
403 if suffix_or_all is None:
404 self.tests.get()[test].keep_all()
405 else:
406 self.tests.get()[test].keep_suffix(suffix_or_all)
407

◆ read()

Optional[Snapshot] pytest_selfie.plugin.SnapshotFileProgress.read ( self,
str test,
str suffix )

Definition at line 422 of file plugin.py.

422 def read(self, test: str, suffix: str) -> Optional[Snapshot]:
423 self.assert_not_terminated()
424 snapshot = self.read_file().snapshots.get(f"{test}{suffix}")
425 if snapshot is not None:
426 self.tests.get()[test].keep_suffix(suffix)
427 return snapshot
428

◆ read_file()

SnapshotFile pytest_selfie.plugin.SnapshotFileProgress.read_file ( self)

Definition at line 429 of file plugin.py.

429 def read_file(self) -> SnapshotFile:
430 if self.file is None:
431 snapshot_path = self.system.layout_pytest.snapshotfile_for_testfile(
432 self.test_file
433 )
434 if os.path.exists(snapshot_path.absolute_path) and os.path.isfile(
435 snapshot_path.absolute_path
436 ):
437 with open(snapshot_path.absolute_path, "rb") as f:
438 content = f.read()
439 self.file = SnapshotFile.parse(SnapshotValueReader.of_binary(content))
440 else:
441 self.file = SnapshotFile.create_empty_with_unix_newlines(
442 self.system.layout_pytest.unix_newlines
443 )
444 return self.file
445
446

◆ test_failed()

pytest_selfie.plugin.SnapshotFileProgress.test_failed ( self,
str testname )

Definition at line 329 of file plugin.py.

329 def test_failed(self, testname: str):
330 self.__assert_in_progress(testname)
331 self.has_failed = True
332 self.tests.get()[testname].keep_all()
333

◆ test_finish()

pytest_selfie.plugin.SnapshotFileProgress.test_finish ( self,
str testname )

Definition at line 334 of file plugin.py.

334 def test_finish(self, testname: str):
335 self.__assert_in_progress(testname)
336 self.finishes_so_far += 1
337 self.testname_in_progress = None
338 if self.finishes_so_far == self.finishes_expected:
339 self.__all_tests_finished()
340

◆ test_start()

pytest_selfie.plugin.SnapshotFileProgress.test_start ( self,
str testname )

Definition at line 318 of file plugin.py.

318 def test_start(self, testname: str):
319 if "/" in testname:
320 raise ValueError(f"Test name cannot contain '/', was {testname}")
321 self.assert_not_terminated()
322 if self.testname_in_progress is not None:
323 raise RuntimeError(
324 f"Cannot start a new test {testname}, {self.testname_in_progress} is already in progress"
325 )
326 self.testname_in_progress = testname
327 self.tests.update_and_get(lambda it: it.plus_or_noop(testname, WithinTestGC()))
328

◆ write()

pytest_selfie.plugin.SnapshotFileProgress.write ( self,
str test,
str suffix,
Snapshot snapshot,
CallStack call_stack,
SnapshotFileLayout layout )

Definition at line 408 of file plugin.py.

415 ):
416 self.assert_not_terminated()
417 key = f"{test}{suffix}"
418 self.disk_write_tracker.record(key, snapshot, call_stack, layout) # type: ignore
419 self.tests.get()[test].keep_suffix(suffix)
420 self.read_file().set_at_test_time(key, snapshot)
421

Member Data Documentation

◆ disk_write_tracker

Optional[DiskWriteTracker] pytest_selfie.plugin.SnapshotFileProgress.disk_write_tracker = DiskWriteTracker()

Definition at line 307 of file plugin.py.

◆ file

Optional[SnapshotFile] pytest_selfie.plugin.SnapshotFileProgress.file = None

Definition at line 303 of file plugin.py.

◆ finishes_expected

int pytest_selfie.plugin.SnapshotFileProgress.finishes_expected = 0

Definition at line 296 of file plugin.py.

◆ finishes_so_far

int pytest_selfie.plugin.SnapshotFileProgress.finishes_so_far = 0

Definition at line 298 of file plugin.py.

◆ has_failed

bool pytest_selfie.plugin.SnapshotFileProgress.has_failed = False

Definition at line 300 of file plugin.py.

◆ system

pytest_selfie.plugin.SnapshotFileProgress.system = system

Definition at line 291 of file plugin.py.

◆ TERMINATED

pytest_selfie.plugin.SnapshotFileProgress.TERMINATED = ArrayMap.empty().plus(" ~ / f!n1shed / ~ ", WithinTestGC())
static

Definition at line 288 of file plugin.py.

◆ test_file

pytest_selfie.plugin.SnapshotFileProgress.test_file = test_file

Definition at line 293 of file plugin.py.

◆ testname_in_progress

Optional[str] pytest_selfie.plugin.SnapshotFileProgress.testname_in_progress = None

Definition at line 309 of file plugin.py.

◆ testname_in_progress_failed

bool pytest_selfie.plugin.SnapshotFileProgress.testname_in_progress_failed = False

Definition at line 310 of file plugin.py.

◆ tests

AtomicReference[ArrayMap[str, WithinTestGC]] pytest_selfie.plugin.SnapshotFileProgress.tests
Initial value:
= AtomicReference(
ArrayMap.empty()
)

Definition at line 304 of file plugin.py.


The documentation for this class was generated from the following file: