1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
|
import importlib.util
import re
import sys
from collections import namedtuple
from pathlib import Path
from textwrap import dedent
from typing import TYPE_CHECKING
from unittest.mock import MagicMock
import pytest
from _pytest.assertion.rewrite import AssertionRewritingHook
from watchfiles.cli import cli
if TYPE_CHECKING:
from conftest import MockRustType
pytestmark = pytest.mark.skipif(sys.platform == 'win32', reason='some tests fail on windows')
ROOT_DIR = Path(__file__).parent.parent
@pytest.fixture
def import_execute(request, tmp_work_path: Path):
def _import_execute(module_name: str, source: str, rewrite_assertions: bool = False):
if rewrite_assertions:
loader = AssertionRewritingHook(config=request.config)
loader.mark_rewrite(module_name)
else:
loader = None
example_bash_file = tmp_work_path / 'example.sh'
example_bash_file.write_text('#!/bin/sh\necho testing')
example_bash_file.chmod(0o755)
(tmp_work_path / 'first/path').mkdir(parents=True, exist_ok=True)
(tmp_work_path / 'second/path').mkdir(parents=True, exist_ok=True)
module_path = tmp_work_path / f'{module_name}.py'
module_path.write_text(source)
spec = importlib.util.spec_from_file_location('__main__', str(module_path), loader=loader)
module = importlib.util.module_from_spec(spec)
try:
spec.loader.exec_module(module)
except KeyboardInterrupt:
print('KeyboardInterrupt')
return _import_execute
def extract_code_chunks(path: Path, text: str, offset: int):
rel_path = path.relative_to(ROOT_DIR)
for m_code in re.finditer(r'^```(.*?)$\n(.*?)^```', text, flags=re.M | re.S):
prefix = m_code.group(1).lower()
if not prefix.startswith(('py', '{.py')):
continue
start_line = offset + text[: m_code.start()].count('\n') + 1
code = m_code.group(2)
end_line = start_line + code.count('\n') + 1
source = '\n' * start_line + code
if 'test="skip"' in prefix:
source = '__skip__'
yield pytest.param(f'{path.stem}_{start_line}_{end_line}', source, id=f'{rel_path}:{start_line}-{end_line}')
def generate_code_chunks(*directories: str):
for d in directories:
for path in (ROOT_DIR / d).glob('**/*'):
if path.suffix == '.py':
code = path.read_text()
for m_docstring in re.finditer(r'(^\s*)r?"""$(.*?)\1"""', code, flags=re.M | re.S):
start_line = code[: m_docstring.start()].count('\n')
docstring = dedent(m_docstring.group(2))
yield from extract_code_chunks(path, docstring, start_line)
elif path.suffix == '.md':
code = path.read_text()
yield from extract_code_chunks(path, code, 0)
# with pypy we sometimes (???) get a "The loop argument is deprecated since Python 3.8" warning, see
# https://github.com/samuelcolvin/watchfiles/runs/7764187741
@pytest.mark.filterwarnings('ignore:The loop argument is deprecated:DeprecationWarning')
@pytest.mark.parametrize('module_name,source_code', generate_code_chunks('watchfiles', 'docs'))
def test_docs_examples(module_name, source_code, import_execute, mocker, mock_rust_notify: 'MockRustType'):
mock_rust_notify([{(1, 'foo.txt'), (2, 'bar.py')}])
mocker.patch('watchfiles.run.spawn_context.Process')
mocker.patch('watchfiles.run.os.kill')
if source_code == '__skip__':
pytest.skip('test="skip" on code snippet')
async def dont_sleep(t):
pass
mocker.patch('asyncio.sleep', new=dont_sleep)
# avoid installing aiohttp by mocking it
sys.modules['aiohttp'] = type('aiohttp', (), {'web': MagicMock()})
try:
import_execute(module_name, source_code, True)
except Exception:
sys.modules.pop('aiohttp', None)
raise
@pytest.mark.skipif(sys.version_info[:2] != (3, 10), reason='output varies between versions')
def test_cli_help(mocker, capsys):
mocker.patch('watchfiles.cli.argparse.ArgumentParser.exit', side_effect=RuntimeError('custom exit'))
TerminalSize = namedtuple('TerminalSize', ['columns', 'lines'])
mocker.patch('shutil.get_terminal_size', return_value=TerminalSize(80, 24))
with pytest.raises(RuntimeError, match='custom exit'):
cli('--help')
out, err = capsys.readouterr()
assert err == ''
cli_help_path = ROOT_DIR / 'docs' / 'cli_help.txt'
try:
assert out == cli_help_path.read_text(), f'cli help output differs from {cli_help_path}, file updated'
except AssertionError:
cli_help_path.write_text(out)
raise
|
