Merge "Ignore the edits unrelated to Android dev in the edit monitor" into main (dbbe1d8a) · Commits · e / os / android_build

tools/edit_monitor/edit_monitor.py

+29 −3

Original line number	Diff line number	Diff line
		@@ -17,6 +17,7 @@ import getpass
		import logging
		import multiprocessing.connection
		import os
		import pathlib
		import platform
		import time

		@@ -65,10 +66,22 @@ class ClearcutEventHandler(PatternMatchingEventHandler):
		def _log_edit_event(
		self, event: FileSystemEvent, edit_type: edit_event_pb2.EditEvent.EditType
		):
		try:
		event_time = time.time()

		if self._is_hidden_file(pathlib.Path(event.src_path)):
		logging.debug("ignore hidden file: %s.", event.src_path)
		return

		if not self._is_under_git_project(pathlib.Path(event.src_path)):
		logging.debug(
		"ignore file %s which does not belong to a git project",
		event.src_path,
		)
		return

		logging.info("%s: %s", event.event_type, event.src_path)
		try:

		event_proto = edit_event_pb2.EditEvent(
		user_name=self.user_name,
		host_name=self.host_name,
		@@ -88,6 +101,19 @@ class ClearcutEventHandler(PatternMatchingEventHandler):
		except Exception:
		logging.exception("Failed to log edit event.")

		def _is_hidden_file(self, file_path: pathlib.Path) -> bool:
		return any(
		part.startswith(".")
		for part in file_path.relative_to(self.root_monitoring_path).parts
		)

		def _is_under_git_project(self, file_path: pathlib.Path) -> bool:
		root_path = pathlib.Path(self.root_monitoring_path).resolve()
		return any(
		root_path.joinpath(dir).joinpath('.git').exists()
		for dir in file_path.relative_to(root_path).parents
		)


		def start(
		path: str,

tools/edit_monitor/edit_monitor_test.py

+67 −0

Original line number	Diff line number	Diff line
		@@ -54,6 +54,8 @@ class EditMonitorTest(unittest.TestCase):
		super().tearDown()

		def test_log_edit_event_success(self):
		# Create the .git file under the monitoring dir.
		self.root_monitoring_path.joinpath('.git').touch()
		fake_cclient = FakeClearcutClient(
		log_output_file=self.log_event_dir.joinpath('logs.output')
		)
		@@ -125,7 +127,72 @@ class EditMonitorTest(unittest.TestCase):
		).single_edit_event,
		)

		def test_do_not_log_edit_event_for_directory_change(self):
		# Create the .git file under the monitoring dir.
		self.root_monitoring_path.joinpath('.git').touch()
		fake_cclient = FakeClearcutClient(
		log_output_file=self.log_event_dir.joinpath('logs.output')
		)
		p = self._start_test_edit_monitor_process(fake_cclient)

		# Create a sub directory
		self.root_monitoring_path.joinpath('test_dir').mkdir()
		# Give some time for the edit monitor to receive the edit event.
		time.sleep(1)
		# Stop the edit monitor and flush all events.
		os.kill(p.pid, signal.SIGINT)
		p.join()

		logged_events = self._get_logged_events()
		self.assertEqual(len(logged_events), 0)

		def test_do_not_log_edit_event_for_hidden_file(self):
		# Create the .git file under the monitoring dir.
		self.root_monitoring_path.joinpath('.git').touch()
		fake_cclient = FakeClearcutClient(
		log_output_file=self.log_event_dir.joinpath('logs.output')
		)
		p = self._start_test_edit_monitor_process(fake_cclient)

		# Create a hidden file.
		self.root_monitoring_path.joinpath('.test.txt').touch()
		# Create a hidden dir.
		hidden_dir = self.root_monitoring_path.joinpath('.test')
		hidden_dir.mkdir()
		hidden_dir.joinpath('test.txt').touch()
		# Give some time for the edit monitor to receive the edit event.
		time.sleep(1)
		# Stop the edit monitor and flush all events.
		os.kill(p.pid, signal.SIGINT)
		p.join()

		logged_events = self._get_logged_events()
		self.assertEqual(len(logged_events), 0)

		def test_do_not_log_edit_event_for_non_git_project_file(self):
		fake_cclient = FakeClearcutClient(
		log_output_file=self.log_event_dir.joinpath('logs.output')
		)
		p = self._start_test_edit_monitor_process(fake_cclient)

		# Create a file.
		self.root_monitoring_path.joinpath('test.txt').touch()
		# Create a file under a sub dir.
		sub_dir = self.root_monitoring_path.joinpath('.test')
		sub_dir.mkdir()
		sub_dir.joinpath('test.txt').touch()
		# Give some time for the edit monitor to receive the edit event.
		time.sleep(1)
		# Stop the edit monitor and flush all events.
		os.kill(p.pid, signal.SIGINT)
		p.join()

		logged_events = self._get_logged_events()
		self.assertEqual(len(logged_events), 0)

		def test_log_edit_event_fail(self):
		# Create the .git file under the monitoring dir.
		self.root_monitoring_path.joinpath('.git').touch()
		fake_cclient = FakeClearcutClient(
		log_output_file=self.log_event_dir.joinpath('logs.output'),
		raise_log_exception=True,