13886086761

Committed 16 Mar 2025 06:05PM UTC coverage: 92.43% (-0.5%) from 92.957%

Build # 13886086761

Build Type

Pull #1156

github

Committed by

web-flow

Commit Message

Merge 2521fddfe into 2b00cbabf

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

Run Details

179 of 205 new or added lines in 3 files covered. (87.32%)

8 existing lines in 1 file now uncovered.

3321 of 3593 relevant lines covered (92.43%)

17.6 hits per line

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

67.39

/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 on_close(self) -> None:
        """Closes the current reporter's output stream"""
        self.current_reporter.on_close(self.linter.stats, None)


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:
        event_handler.on_close()
        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:
31	self.files_to_watch = set(files_to_watch)	20✔
32	self.linter = linter	20✔
33	self.current_reporter = current_reporter	20✔
34	self.local_config = local_config	20✔
35	self.load_default_config = load_default_config	20✔
36	self.autoformat = autoformat	20✔
37	self.level = level	20✔
38	self.f_paths = f_paths	20✔
39
40	def on_modified(self, event) -> None:	20✔
41	"""Trigger the callback when a watched file is modified."""
42	if event.src_path not in self.files_to_watch:	20✔
43	return	20✔
44
45	logging.info(f"File modified: {event.src_path}, re-running checks...")	20✔
46	if event.src_path in self.current_reporter.messages:	20✔
NEW 47	del self.current_reporter.messages[event.src_path]	×
48
49	_, self.linter = check_file(	20✔
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	)
58	self.current_reporter = self.linter.reporter	20✔
59	self.current_reporter.print_messages(self.level)	20✔
60	self.linter.generate_reports()	20✔
61	upload_linter_results(self.linter, self.current_reporter, self.f_paths, self.local_config)	20✔
62
63	def on_close(self) -> None:	20✔
64	"""Closes the current reporter's output stream"""
NEW 65	self.current_reporter.on_close(self.linter.stats, None)	×
66
67
68	def watch_files(	20✔
69	file_paths: set,
70	level: str,
71	local_config: Union[dict[str, Any], str],
72	load_default_config: bool,
73	autoformat: Optional[bool],
74	linter: PyLinter,
75	current_reporter: Union[BaseReporter, MultiReporter],
76	f_paths: list[str],
77	) -> None:
78	"""Watch a list of files for modifications and trigger a callback when changes occur."""
NEW 79	logging.info("PythonTA is monitoring your files for changes and will re-check after every save")	×
NEW 80	directories_to_watch = {os.path.dirname(file) for file in file_paths}	×
NEW 81	event_handler = FileChangeHandler(	×
82	files_to_watch=file_paths,
83	linter=linter,
84	current_reporter=current_reporter,
85	local_config=local_config,
86	load_default_config=load_default_config,
87	autoformat=autoformat,
88	level=level,
89	f_paths=f_paths,
90	)
NEW 91	observer = Observer()	×
NEW 92	for directory in directories_to_watch:	×
NEW 93	observer.schedule(event_handler, path=directory, recursive=False)	×
NEW 94	observer.start()	×
95
NEW 96	try:	×
97	while True:
NEW 98	time.sleep(1)	×
NEW 99	except KeyboardInterrupt:	×
NEW 100	event_handler.on_close()	×
NEW 101	observer.stop()	×
102
NEW 103	observer.join()	×

pyta-uoft / pyta / 13886086761

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