Coverage for tests / test_atomic_file.py: 100%

1import subprocess 7 ctx1abcdefg

2import sys 7 ctx1abcdefg

3from pathlib import Path 7 ctx1abcdefg

4 

5import pytest 7 ctx1abcdefg

6 

7from . import atomic_write_example as mod 7 ctx1abcdefg

8 

9 

10def test_atomic_write(tmp_path: Path) -> None: 7 ctx1abcdefg

11 original_content = "existing-content\n" 7 ctx1abcdefg

12 output_file = tmp_path / "atomic-write-target.txt" 7 ctx1abcdefg

13 output_file.write_text(original_content, encoding="utf-8") 7 ctx1abcdefg

14 

15 process = subprocess.Popen( 7 ctx1abcdefg

16 [ 

17 sys.executable, 

18 "-m", 

19 "coverage", 

20 "run", 

21 mod.__file__, 

22 "write-atomic", 

23 f"--config={output_file}", 

24 ], 

25 stdout=subprocess.PIPE, 

26 stderr=subprocess.PIPE, 

27 text=True, 

28 ) 

29 assert process.stdout is not None 7 ctx1abcdefg

30 

31 # Halfway of writing the file, check that the original content is still there 

32 halfway_line = process.stdout.readline().strip() 7 ctx1abcdefg

33 assert halfway_line == "halfway" 7 ctx1abcdefg

34 assert output_file.read_text(encoding="utf-8") == original_content 7 ctx1abcdefg

35 

36 # Only at the end, the full new content is visible 

37 stdout, stderr = process.communicate(timeout=5) 7 ctx1abcdefg

38 assert process.returncode == 0, stderr 7 ctx1abcdefg

39 assert "written atomically" in stdout 7 ctx1abcdefg

40 assert ( 7 ctx1abcdefg

41 output_file.read_text(encoding="utf-8") 

42 == "atomic-content-1\natomic-content-2\n" 

43 ) 

44 

45 

46def test_atomic_binary_write(tmp_path: Path) -> None: 7 ctx1abcdefg

47 output_file = tmp_path / "atomic-binary.bin" 7 ctx1abcdefg

48 

49 result = subprocess.run( 7 ctx1abcdefg

50 [ 

51 sys.executable, 

52 "-m", 

53 "coverage", 

54 "run", 

55 mod.__file__, 

56 "write-atomic-binary", 

57 f"--config={output_file}", 

58 ], 

59 capture_output=True, 

60 encoding="utf-8", 

61 ) 

62 

63 assert result.returncode == 0, result.stderr 7 ctx1abcdefg

64 assert "written binary atomically" in result.stdout 7 ctx1abcdefg

65 assert output_file.read_bytes() == b"\x00\x01binary-atomic\n" 7 ctx1abcdefg

66 

67 

68def test_atomic_api(tmp_path: Path) -> None: 7 ctx1abcdefg

69 output_file = tmp_path / "atomic-api.txt" 7 ctx1abcdefg

70 

71 result = subprocess.run( 7 ctx1abcdefg

72 [ 

73 sys.executable, 

74 "-m", 

75 "coverage", 

76 "run", 

77 mod.__file__, 

78 "api-atomic", 

79 f"--config={output_file}", 

80 ], 

81 capture_output=True, 

82 encoding="utf-8", 

83 ) 

84 

85 assert result.returncode == 0, result.stderr 7 ctx1abcdefg

86 assert f"name={output_file}" in result.stdout 7 ctx1abcdefg

87 assert "repr=<_io.TextIOWrapper" in result.stdout 7 ctx1abcdefg

88 assert "entered=True" in result.stdout 7 ctx1abcdefg

89 assert output_file.read_text(encoding="utf-8") == "atomic-api-done\n" 7 ctx1abcdefg

90 

91 

92@pytest.mark.parametrize( 7 ctx1abcdefg

93 ("command_name", "expected_message"), 

94 [ 

95 ("invalid-atomic-append", "Appending to an existing file is not supported"), 

96 ("invalid-atomic-exclusive", "Use the `overwrite`-parameter instead."), 

97 ("invalid-atomic-read", "Atomic writes only make sense with `w`-mode."), 

98 ], 

99) 

100def test_atomic_mode_invalid_options( 7 ctx1abcdefg

101 tmp_path: Path, command_name: str, expected_message: str 

102) -> None: 

103 output_file = tmp_path / "atomic-invalid-mode.txt" 7 ctx1abcdefg

104 output_file.write_text("existing-content\n", encoding="utf-8") 7 ctx1abcdefg

105 

106 result = subprocess.run( 7 ctx1abcdefg

107 [ 

108 sys.executable, 

109 "-m", 

110 "coverage", 

111 "run", 

112 mod.__file__, 

113 command_name, 

114 f"--config={output_file}", 

115 ], 

116 capture_output=True, 

117 encoding="utf-8", 

118 ) 

119 

120 assert result.returncode != 0 7 ctx1abcdefg

121 combined_output = f"{result.stdout}\n{result.stderr}" 7 ctx1abcdefg

122 assert expected_message in combined_output 7 ctx1abcdefg