13801531304

Committed 10 Mar 2025 03:51AM UTC coverage: 92.031% (-0.9%) from 92.957%

Build # 13801531304

Build Type

Pull #1156

github

Committed by

web-flow

Commit Message

Merge 589e90852 into 2b00cbabf

Pull Request Pull Request #1156: Integrate Watchdog for Live Code Re-Checking

Run Details

160 of 200 new or added lines in 3 files covered. (80.0%)

8 existing lines in 1 file now uncovered.

3303 of 3589 relevant lines covered (92.03%)

17.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

30.23

/python_ta/check/watch.py

"""Module to watch files for modifications and trigger PythonTA checks automatically.
"""

import logging
import os
import time
from typing import Any, Optional, Union

from pylint.lint import PyLinter
from pylint.reporters import BaseReporter, MultiReporter
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

from .helpers import check_file, upload_linter_results


class FileChangeHandler(FileSystemEventHandler):
    """Internal class to handle file modifications."""

    def __init__(
        self,
        files_to_watch: set,
        linter: PyLinter,
        current_reporter: Union[BaseReporter, MultiReporter],
        local_config: Union[dict[str, Any], str],
        load_default_config: bool,
        autoformat: Optional[bool],
        level: str,
        f_paths: list[str],
    ) -> None:
        self.files_to_watch = set(files_to_watch)
        self.linter = linter
        self.current_reporter = current_reporter
        self.local_config = local_config
        self.load_default_config = load_default_config
        self.autoformat = autoformat
        self.level = level
        self.f_paths = f_paths

    def on_modified(self, event) -> None:
        """Trigger the callback when a watched file is modified."""
        if event.src_path not in self.files_to_watch:
            return

        logging.info(f"File modified: {event.src_path}, re-running checks...")
        if event.src_path in self.current_reporter.messages:
            del self.current_reporter.messages[event.src_path]

        _, self.linter = check_file(
            file_py=event.src_path,
            local_config=self.local_config,
            load_default_config=self.load_default_config,
            autoformat=self.autoformat,
            is_any_file_checked=True,
            current_reporter=self.current_reporter,
            f_paths=[],
        )
        self.current_reporter = self.linter.reporter
        self.current_reporter.print_messages(self.level)
        self.linter.generate_reports()
        upload_linter_results(self.linter, self.current_reporter, self.f_paths, self.local_config)


def watch_files(
    file_paths: set,
    level: str,
    local_config: Union[dict[str, Any], str],
    load_default_config: bool,
    autoformat: Optional[bool],
    linter: PyLinter,
    current_reporter: Union[BaseReporter, MultiReporter],
    f_paths: list[str],
) -> None:
    """Watch a list of files for modifications and trigger a callback when changes occur."""
    logging.info("PythonTA is monitoring your files for changes and will re-check after every save")
    directories_to_watch = {os.path.dirname(file) for file in file_paths}
    event_handler = FileChangeHandler(
        files_to_watch=file_paths,
        linter=linter,
        current_reporter=current_reporter,
        local_config=local_config,
        load_default_config=load_default_config,
        autoformat=autoformat,
        level=level,
        f_paths=f_paths,
    )
    observer = Observer()
    for directory in directories_to_watch:
        observer.schedule(event_handler, path=directory, recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

1	"""Module to watch files for modifications and trigger PythonTA checks automatically.
2	"""
3
4	import logging	20✔
5	import os	20✔
6	import time	20✔
7	from typing import Any, Optional, Union	20✔
8
9	from pylint.lint import PyLinter	20✔
10	from pylint.reporters import BaseReporter, MultiReporter	20✔
11	from watchdog.events import FileSystemEventHandler	20✔
12	from watchdog.observers import Observer	20✔
13
14	from .helpers import check_file, upload_linter_results	20✔
15
16
17	class FileChangeHandler(FileSystemEventHandler):	20✔
18	"""Internal class to handle file modifications."""
19
20	def __init__(	20✔
21	self,
22	files_to_watch: set,
23	linter: PyLinter,
24	current_reporter: Union[BaseReporter, MultiReporter],
25	local_config: Union[dict[str, Any], str],
26	load_default_config: bool,
27	autoformat: Optional[bool],
28	level: str,
29	f_paths: list[str],
30	) -> None:
NEW 31	self.files_to_watch = set(files_to_watch)	×
NEW 32	self.linter = linter	×
NEW 33	self.current_reporter = current_reporter	×
NEW 34	self.local_config = local_config	×
NEW 35	self.load_default_config = load_default_config	×
NEW 36	self.autoformat = autoformat	×
NEW 37	self.level = level	×
NEW 38	self.f_paths = f_paths	×
39
40	def on_modified(self, event) -> None:	20✔
41	"""Trigger the callback when a watched file is modified."""
NEW 42	if event.src_path not in self.files_to_watch:	×
NEW 43	return	×
44
NEW 45	logging.info(f"File modified: {event.src_path}, re-running checks...")	×
NEW 46	if event.src_path in self.current_reporter.messages:	×
NEW 47	del self.current_reporter.messages[event.src_path]	×
48
NEW 49	_, self.linter = check_file(	×
50	file_py=event.src_path,
51	local_config=self.local_config,
52	load_default_config=self.load_default_config,
53	autoformat=self.autoformat,
54	is_any_file_checked=True,
55	current_reporter=self.current_reporter,
56	f_paths=[],
57	)
NEW 58	self.current_reporter = self.linter.reporter	×
NEW 59	self.current_reporter.print_messages(self.level)	×
NEW 60	self.linter.generate_reports()	×
NEW 61	upload_linter_results(self.linter, self.current_reporter, self.f_paths, self.local_config)	×
62
63
64	def watch_files(	20✔
65	file_paths: set,
66	level: str,
67	local_config: Union[dict[str, Any], str],
68	load_default_config: bool,
69	autoformat: Optional[bool],
70	linter: PyLinter,
71	current_reporter: Union[BaseReporter, MultiReporter],
72	f_paths: list[str],
73	) -> None:
74	"""Watch a list of files for modifications and trigger a callback when changes occur."""
NEW 75	logging.info("PythonTA is monitoring your files for changes and will re-check after every save")	×
NEW 76	directories_to_watch = {os.path.dirname(file) for file in file_paths}	×
NEW 77	event_handler = FileChangeHandler(	×
78	files_to_watch=file_paths,
79	linter=linter,
80	current_reporter=current_reporter,
81	local_config=local_config,
82	load_default_config=load_default_config,
83	autoformat=autoformat,
84	level=level,
85	f_paths=f_paths,
86	)
NEW 87	observer = Observer()	×
NEW 88	for directory in directories_to_watch:	×
NEW 89	observer.schedule(event_handler, path=directory, recursive=False)	×
NEW 90	observer.start()	×
91
NEW 92	try:	×
93	while True:
NEW 94	time.sleep(1)	×
NEW 95	except KeyboardInterrupt:	×
NEW 96	observer.stop()	×
97
NEW 98	observer.join()	×

pyta-uoft / pyta / 13801531304

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous