1 files changed, 373 insertions, 352 deletions
diff --git a/tests/test_git_trace2_event_log.py b/tests/test_git_trace2_event_log.py
index 7e7dfb7a..a6078d38 100644
--- a/tests/test_git_trace2_event_log.py
+++ b/tests/test_git_trace2_event_log.py
@@ -27,361 +27,382 @@ import platform_utils
 def serverLoggingThread(socket_path, server_ready, received_traces):
-  """Helper function to receive logs over a Unix domain socket.
+    """Helper function to receive logs over a Unix domain socket.
-  Appends received messages on the provided socket and appends to received_traces.
-  Args:
-    socket_path: path to a Unix domain socket on which to listen for traces
-    server_ready: a threading.Condition used to signal to the caller that this thread is ready to
-        accept connections
-    received_traces: a list to which received traces will be appended (after decoding to a utf-8
-        string).
-  """
-  platform_utils.remove(socket_path, missing_ok=True)
-  data = b''
-  with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
-    sock.bind(socket_path)
-    sock.listen(0)
-    with server_ready:
-      server_ready.notify()
-    with sock.accept()[0] as conn:
-      while True:
-        recved = conn.recv(4096)
-        if not recved:
-          break
-        data += recved
-  received_traces.extend(data.decode('utf-8').splitlines())
+    Appends received messages on the provided socket and appends to
+    received_traces.
-class EventLogTestCase(unittest.TestCase):
+    Args:
-  """TestCase for the EventLog module."""
+        socket_path: path to a Unix domain socket on which to listen for traces
+        server_ready: a threading.Condition used to signal to the caller that
-  PARENT_SID_KEY = 'GIT_TRACE2_PARENT_SID'
+            this thread is ready to accept connections
-  PARENT_SID_VALUE = 'parent_sid'
+        received_traces: a list to which received traces will be appended (after
-  SELF_SID_REGEX = r'repo-\d+T\d+Z-.*'
+            decoding to a utf-8 string).
-  FULL_SID_REGEX = r'^%s/%s' % (PARENT_SID_VALUE, SELF_SID_REGEX)
-  def setUp(self):
-    """Load the event_log module every time."""
-    self._event_log_module = None
-    # By default we initialize with the expected case where
-    # repo launches us (so GIT_TRACE2_PARENT_SID is set).
-    env = {
-        self.PARENT_SID_KEY: self.PARENT_SID_VALUE,
-    }
-    self._event_log_module = git_trace2_event_log.EventLog(env=env)
-    self._log_data = None
-  def verifyCommonKeys(self, log_entry, expected_event_name=None, full_sid=True):
-    """Helper function to verify common event log keys."""
-    self.assertIn('event', log_entry)
-    self.assertIn('sid', log_entry)
-    self.assertIn('thread', log_entry)
-    self.assertIn('time', log_entry)
-    # Do basic data format validation.
-    if expected_event_name:
-      self.assertEqual(expected_event_name, log_entry['event'])
-    if full_sid:
-      self.assertRegex(log_entry['sid'], self.FULL_SID_REGEX)
-    else:
-      self.assertRegex(log_entry['sid'], self.SELF_SID_REGEX)
-    self.assertRegex(log_entry['time'], r'^\d+-\d+-\d+T\d+:\d+:\d+\.\d+Z$')
-  def readLog(self, log_path):
-    """Helper function to read log data into a list."""
-    log_data = []
-    with open(log_path, mode='rb') as f:
-      for line in f:
-        log_data.append(json.loads(line))
-    return log_data
-  def remove_prefix(self, s, prefix):
-    """Return a copy string after removing |prefix| from |s|, if present or the original string."""
-    if s.startswith(prefix):
-        return s[len(prefix):]
-    else:
-        return s
-  def test_initial_state_with_parent_sid(self):
-    """Test initial state when 'GIT_TRACE2_PARENT_SID' is set by parent."""
-    self.assertRegex(self._event_log_module.full_sid, self.FULL_SID_REGEX)
-  def test_initial_state_no_parent_sid(self):
-    """Test initial state when 'GIT_TRACE2_PARENT_SID' is not set."""
-    # Setup an empty environment dict (no parent sid).
-    self._event_log_module = git_trace2_event_log.EventLog(env={})
-    self.assertRegex(self._event_log_module.full_sid, self.SELF_SID_REGEX)
-  def test_version_event(self):
-    """Test 'version' event data is valid.
-    Verify that the 'version' event is written even when no other
-    events are addded.
-    Expected event log:
-    <version event>
-    """
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    # A log with no added events should only have the version entry.
-    self.assertEqual(len(self._log_data), 1)
-    version_event = self._log_data[0]
-    self.verifyCommonKeys(version_event, expected_event_name='version')
-    # Check for 'version' event specific fields.
-    self.assertIn('evt', version_event)
-    self.assertIn('exe', version_event)
-    # Verify "evt" version field is a string.
-    self.assertIsInstance(version_event['evt'], str)
-  def test_start_event(self):
-    """Test and validate 'start' event data is valid.
-    Expected event log:
-    <version event>
-    <start event>
-    """
-    self._event_log_module.StartEvent()
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    self.assertEqual(len(self._log_data), 2)
-    start_event = self._log_data[1]
-    self.verifyCommonKeys(self._log_data[0], expected_event_name='version')
-    self.verifyCommonKeys(start_event, expected_event_name='start')
-    # Check for 'start' event specific fields.
-    self.assertIn('argv', start_event)
-    self.assertTrue(isinstance(start_event['argv'], list))
-  def test_exit_event_result_none(self):
-    """Test 'exit' event data is valid when result is None.
-    We expect None result to be converted to 0 in the exit event data.
-    Expected event log:
-    <version event>
-    <exit event>
-    """
-    self._event_log_module.ExitEvent(None)
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    self.assertEqual(len(self._log_data), 2)
-    exit_event = self._log_data[1]
-    self.verifyCommonKeys(self._log_data[0], expected_event_name='version')
-    self.verifyCommonKeys(exit_event, expected_event_name='exit')
-    # Check for 'exit' event specific fields.
-    self.assertIn('code', exit_event)
-    # 'None' result should convert to 0 (successful) return code.
-    self.assertEqual(exit_event['code'], 0)
-  def test_exit_event_result_integer(self):
-    """Test 'exit' event data is valid when result is an integer.
-    Expected event log:
-    <version event>
-    <exit event>
-    """
-    self._event_log_module.ExitEvent(2)
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    self.assertEqual(len(self._log_data), 2)
-    exit_event = self._log_data[1]
-    self.verifyCommonKeys(self._log_data[0], expected_event_name='version')
-    self.verifyCommonKeys(exit_event, expected_event_name='exit')
-    # Check for 'exit' event specific fields.
-    self.assertIn('code', exit_event)
-    self.assertEqual(exit_event['code'], 2)
-  def test_command_event(self):
-    """Test and validate 'command' event data is valid.
-    Expected event log:
-    <version event>
-    <command event>
-    """
-    name = 'repo'
-    subcommands = ['init' 'this']
-    self._event_log_module.CommandEvent(name='repo', subcommands=subcommands)
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    self.assertEqual(len(self._log_data), 2)
-    command_event = self._log_data[1]
-    self.verifyCommonKeys(self._log_data[0], expected_event_name='version')
-    self.verifyCommonKeys(command_event, expected_event_name='command')
-    # Check for 'command' event specific fields.
-    self.assertIn('name', command_event)
-    self.assertIn('subcommands', command_event)
-    self.assertEqual(command_event['name'], name)
-    self.assertEqual(command_event['subcommands'], subcommands)
-  def test_def_params_event_repo_config(self):
-    """Test 'def_params' event data outputs only repo config keys.
-    Expected event log:
-    <version event>
-    <def_param event>
-    <def_param event>
    """
-    config = {
+    platform_utils.remove(socket_path, missing_ok=True)
-        'git.foo': 'bar',
+    data = b""
-        'repo.partialclone': 'true',
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
-        'repo.partialclonefilter': 'blob:none',
+        sock.bind(socket_path)
-    }
+        sock.listen(0)
-    self._event_log_module.DefParamRepoEvents(config)
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    self.assertEqual(len(self._log_data), 3)
-    def_param_events = self._log_data[1:]
-    self.verifyCommonKeys(self._log_data[0], expected_event_name='version')
-    for event in def_param_events:
-      self.verifyCommonKeys(event, expected_event_name='def_param')
-      # Check for 'def_param' event specific fields.
-      self.assertIn('param', event)
-      self.assertIn('value', event)
-      self.assertTrue(event['param'].startswith('repo.'))
-  def test_def_params_event_no_repo_config(self):
-    """Test 'def_params' event data won't output non-repo config keys.
-    Expected event log:
-    <version event>
-    """
-    config = {
-        'git.foo': 'bar',
-        'git.core.foo2': 'baz',
-    }
-    self._event_log_module.DefParamRepoEvents(config)
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    self.assertEqual(len(self._log_data), 1)
-    self.verifyCommonKeys(self._log_data[0], expected_event_name='version')
-  def test_data_event_config(self):
-    """Test 'data' event data outputs all config keys.
-    Expected event log:
-    <version event>
-    <data event>
-    <data event>
-    """
-    config = {
-        'git.foo': 'bar',
-        'repo.partialclone': 'false',
-        'repo.syncstate.superproject.hassuperprojecttag': 'true',
-        'repo.syncstate.superproject.sys.argv': ['--', 'sync', 'protobuf'],
-    }
-    prefix_value = 'prefix'
-    self._event_log_module.LogDataConfigEvents(config, prefix_value)
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    self.assertEqual(len(self._log_data), 5)
-    data_events = self._log_data[1:]
-    self.verifyCommonKeys(self._log_data[0], expected_event_name='version')
-    for event in data_events:
-      self.verifyCommonKeys(event)
-      # Check for 'data' event specific fields.
-      self.assertIn('key', event)
-      self.assertIn('value', event)
-      key = event['key']
-      key = self.remove_prefix(key, f'{prefix_value}/')
-      value = event['value']
-      self.assertEqual(self._event_log_module.GetDataEventName(value), event['event'])
-      self.assertTrue(key in config and value == config[key])
-  def test_error_event(self):
-    """Test and validate 'error' event data is valid.
-    Expected event log:
-    <version event>
-    <error event>
-    """
-    msg = 'invalid option: --cahced'
-    fmt = 'invalid option: %s'
-    self._event_log_module.ErrorEvent(msg, fmt)
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      log_path = self._event_log_module.Write(path=tempdir)
-      self._log_data = self.readLog(log_path)
-    self.assertEqual(len(self._log_data), 2)
-    error_event = self._log_data[1]
-    self.verifyCommonKeys(self._log_data[0], expected_event_name='version')
-    self.verifyCommonKeys(error_event, expected_event_name='error')
-    # Check for 'error' event specific fields.
-    self.assertIn('msg', error_event)
-    self.assertIn('fmt', error_event)
-    self.assertEqual(error_event['msg'], msg)
-    self.assertEqual(error_event['fmt'], fmt)
-  def test_write_with_filename(self):
-    """Test Write() with a path to a file exits with None."""
-    self.assertIsNone(self._event_log_module.Write(path='path/to/file'))
-  def test_write_with_git_config(self):
-    """Test Write() uses the git config path when 'git config' call succeeds."""
-    with tempfile.TemporaryDirectory(prefix='event_log_tests') as tempdir:
-      with mock.patch.object(self._event_log_module,
-                             '_GetEventTargetPath', return_value=tempdir):
-        self.assertEqual(os.path.dirname(self._event_log_module.Write()), tempdir)
-  def test_write_no_git_config(self):
-    """Test Write() with no git config variable present exits with None."""
-    with mock.patch.object(self._event_log_module,
-                           '_GetEventTargetPath', return_value=None):
-      self.assertIsNone(self._event_log_module.Write())
-  def test_write_non_string(self):
-    """Test Write() with non-string type for |path| throws TypeError."""
-    with self.assertRaises(TypeError):
-      self._event_log_module.Write(path=1234)
-  def test_write_socket(self):
-    """Test Write() with Unix domain socket for |path| and validate received traces."""
-    received_traces = []
-    with tempfile.TemporaryDirectory(prefix='test_server_sockets') as tempdir:
-      socket_path = os.path.join(tempdir, "server.sock")
-      server_ready = threading.Condition()
-      # Start "server" listening on Unix domain socket at socket_path.
-      try:
-        server_thread = threading.Thread(
-            target=serverLoggingThread,
-            args=(socket_path, server_ready, received_traces))
-        server_thread.start()
        with server_ready:
-          server_ready.wait(timeout=120)
+            server_ready.notify()
+        with sock.accept()[0] as conn:
+            while True:
+                recved = conn.recv(4096)
+                if not recved:
+                    break
+                data += recved
+    received_traces.extend(data.decode("utf-8").splitlines())
+class EventLogTestCase(unittest.TestCase):
+    """TestCase for the EventLog module."""
+    PARENT_SID_KEY = "GIT_TRACE2_PARENT_SID"
+    PARENT_SID_VALUE = "parent_sid"
+    SELF_SID_REGEX = r"repo-\d+T\d+Z-.*"
+    FULL_SID_REGEX = r"^%s/%s" % (PARENT_SID_VALUE, SELF_SID_REGEX)
+    def setUp(self):
+        """Load the event_log module every time."""
+        self._event_log_module = None
+        # By default we initialize with the expected case where
+        # repo launches us (so GIT_TRACE2_PARENT_SID is set).
+        env = {
+            self.PARENT_SID_KEY: self.PARENT_SID_VALUE,
+        }
+        self._event_log_module = git_trace2_event_log.EventLog(env=env)
+        self._log_data = None
+    def verifyCommonKeys(
+        self, log_entry, expected_event_name=None, full_sid=True
+    ):
+        """Helper function to verify common event log keys."""
+        self.assertIn("event", log_entry)
+        self.assertIn("sid", log_entry)
+        self.assertIn("thread", log_entry)
+        self.assertIn("time", log_entry)
+        # Do basic data format validation.
+        if expected_event_name:
+            self.assertEqual(expected_event_name, log_entry["event"])
+        if full_sid:
+            self.assertRegex(log_entry["sid"], self.FULL_SID_REGEX)
+        else:
+            self.assertRegex(log_entry["sid"], self.SELF_SID_REGEX)
+        self.assertRegex(log_entry["time"], r"^\d+-\d+-\d+T\d+:\d+:\d+\.\d+Z$")
+    def readLog(self, log_path):
+        """Helper function to read log data into a list."""
+        log_data = []
+        with open(log_path, mode="rb") as f:
+            for line in f:
+                log_data.append(json.loads(line))
+        return log_data
+    def remove_prefix(self, s, prefix):
+        """Return a copy string after removing |prefix| from |s|, if present or
+        the original string."""
+        if s.startswith(prefix):
+            return s[len(prefix) :]
+        else:
+            return s
+    def test_initial_state_with_parent_sid(self):
+        """Test initial state when 'GIT_TRACE2_PARENT_SID' is set by parent."""
+        self.assertRegex(self._event_log_module.full_sid, self.FULL_SID_REGEX)
+    def test_initial_state_no_parent_sid(self):
+        """Test initial state when 'GIT_TRACE2_PARENT_SID' is not set."""
+        # Setup an empty environment dict (no parent sid).
+        self._event_log_module = git_trace2_event_log.EventLog(env={})
+        self.assertRegex(self._event_log_module.full_sid, self.SELF_SID_REGEX)
+    def test_version_event(self):
+        """Test 'version' event data is valid.
+        Verify that the 'version' event is written even when no other
+        events are addded.
+        Expected event log:
+        <version event>
+        """
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            log_path = self._event_log_module.Write(path=tempdir)
+            self._log_data = self.readLog(log_path)
+        # A log with no added events should only have the version entry.
+        self.assertEqual(len(self._log_data), 1)
+        version_event = self._log_data[0]
+        self.verifyCommonKeys(version_event, expected_event_name="version")
+        # Check for 'version' event specific fields.
+        self.assertIn("evt", version_event)
+        self.assertIn("exe", version_event)
+        # Verify "evt" version field is a string.
+        self.assertIsInstance(version_event["evt"], str)
+    def test_start_event(self):
+        """Test and validate 'start' event data is valid.
+        Expected event log:
+        <version event>
+        <start event>
+        """
        self._event_log_module.StartEvent()
-        path = self._event_log_module.Write(path=f'af_unix:{socket_path}')
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
-      finally:
+            log_path = self._event_log_module.Write(path=tempdir)
-        server_thread.join(timeout=5)
+            self._log_data = self.readLog(log_path)
-    self.assertEqual(path, f'af_unix:stream:{socket_path}')
+        self.assertEqual(len(self._log_data), 2)
-    self.assertEqual(len(received_traces), 2)
+        start_event = self._log_data[1]
-    version_event = json.loads(received_traces[0])
+        self.verifyCommonKeys(self._log_data[0], expected_event_name="version")
-    start_event = json.loads(received_traces[1])
+        self.verifyCommonKeys(start_event, expected_event_name="start")
-    self.verifyCommonKeys(version_event, expected_event_name='version')
+        # Check for 'start' event specific fields.
-    self.verifyCommonKeys(start_event, expected_event_name='start')
+        self.assertIn("argv", start_event)
-    # Check for 'start' event specific fields.
+        self.assertTrue(isinstance(start_event["argv"], list))
-    self.assertIn('argv', start_event)
-    self.assertIsInstance(start_event['argv'], list)
+    def test_exit_event_result_none(self):
+        """Test 'exit' event data is valid when result is None.
+        We expect None result to be converted to 0 in the exit event data.
+        Expected event log:
+        <version event>
+        <exit event>
+        """
+        self._event_log_module.ExitEvent(None)
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            log_path = self._event_log_module.Write(path=tempdir)
+            self._log_data = self.readLog(log_path)
+        self.assertEqual(len(self._log_data), 2)
+        exit_event = self._log_data[1]
+        self.verifyCommonKeys(self._log_data[0], expected_event_name="version")
+        self.verifyCommonKeys(exit_event, expected_event_name="exit")
+        # Check for 'exit' event specific fields.
+        self.assertIn("code", exit_event)
+        # 'None' result should convert to 0 (successful) return code.
+        self.assertEqual(exit_event["code"], 0)
+    def test_exit_event_result_integer(self):
+        """Test 'exit' event data is valid when result is an integer.
+        Expected event log:
+        <version event>
+        <exit event>
+        """
+        self._event_log_module.ExitEvent(2)
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            log_path = self._event_log_module.Write(path=tempdir)
+            self._log_data = self.readLog(log_path)
+        self.assertEqual(len(self._log_data), 2)
+        exit_event = self._log_data[1]
+        self.verifyCommonKeys(self._log_data[0], expected_event_name="version")
+        self.verifyCommonKeys(exit_event, expected_event_name="exit")
+        # Check for 'exit' event specific fields.
+        self.assertIn("code", exit_event)
+        self.assertEqual(exit_event["code"], 2)
+    def test_command_event(self):
+        """Test and validate 'command' event data is valid.
+        Expected event log:
+        <version event>
+        <command event>
+        """
+        name = "repo"
+        subcommands = ["init" "this"]
+        self._event_log_module.CommandEvent(
+            name="repo", subcommands=subcommands
+        )
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            log_path = self._event_log_module.Write(path=tempdir)
+            self._log_data = self.readLog(log_path)
+        self.assertEqual(len(self._log_data), 2)
+        command_event = self._log_data[1]
+        self.verifyCommonKeys(self._log_data[0], expected_event_name="version")
+        self.verifyCommonKeys(command_event, expected_event_name="command")
+        # Check for 'command' event specific fields.
+        self.assertIn("name", command_event)
+        self.assertIn("subcommands", command_event)
+        self.assertEqual(command_event["name"], name)
+        self.assertEqual(command_event["subcommands"], subcommands)
+    def test_def_params_event_repo_config(self):
+        """Test 'def_params' event data outputs only repo config keys.
+        Expected event log:
+        <version event>
+        <def_param event>
+        <def_param event>
+        """
+        config = {
+            "git.foo": "bar",
+            "repo.partialclone": "true",
+            "repo.partialclonefilter": "blob:none",
+        }
+        self._event_log_module.DefParamRepoEvents(config)
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            log_path = self._event_log_module.Write(path=tempdir)
+            self._log_data = self.readLog(log_path)
+        self.assertEqual(len(self._log_data), 3)
+        def_param_events = self._log_data[1:]
+        self.verifyCommonKeys(self._log_data[0], expected_event_name="version")
+        for event in def_param_events:
+            self.verifyCommonKeys(event, expected_event_name="def_param")
+            # Check for 'def_param' event specific fields.
+            self.assertIn("param", event)
+            self.assertIn("value", event)
+            self.assertTrue(event["param"].startswith("repo."))
+    def test_def_params_event_no_repo_config(self):
+        """Test 'def_params' event data won't output non-repo config keys.
+        Expected event log:
+        <version event>
+        """
+        config = {
+            "git.foo": "bar",
+            "git.core.foo2": "baz",
+        }
+        self._event_log_module.DefParamRepoEvents(config)
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            log_path = self._event_log_module.Write(path=tempdir)
+            self._log_data = self.readLog(log_path)
+        self.assertEqual(len(self._log_data), 1)
+        self.verifyCommonKeys(self._log_data[0], expected_event_name="version")
+    def test_data_event_config(self):
+        """Test 'data' event data outputs all config keys.
+        Expected event log:
+        <version event>
+        <data event>
+        <data event>
+        """
+        config = {
+            "git.foo": "bar",
+            "repo.partialclone": "false",
+            "repo.syncstate.superproject.hassuperprojecttag": "true",
+            "repo.syncstate.superproject.sys.argv": ["--", "sync", "protobuf"],
+        }
+        prefix_value = "prefix"
+        self._event_log_module.LogDataConfigEvents(config, prefix_value)
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            log_path = self._event_log_module.Write(path=tempdir)
+            self._log_data = self.readLog(log_path)
+        self.assertEqual(len(self._log_data), 5)
+        data_events = self._log_data[1:]
+        self.verifyCommonKeys(self._log_data[0], expected_event_name="version")
+        for event in data_events:
+            self.verifyCommonKeys(event)
+            # Check for 'data' event specific fields.
+            self.assertIn("key", event)
+            self.assertIn("value", event)
+            key = event["key"]
+            key = self.remove_prefix(key, f"{prefix_value}/")
+            value = event["value"]
+            self.assertEqual(
+                self._event_log_module.GetDataEventName(value), event["event"]
+            )
+            self.assertTrue(key in config and value == config[key])
+    def test_error_event(self):
+        """Test and validate 'error' event data is valid.
+        Expected event log:
+        <version event>
+        <error event>
+        """
+        msg = "invalid option: --cahced"
+        fmt = "invalid option: %s"
+        self._event_log_module.ErrorEvent(msg, fmt)
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            log_path = self._event_log_module.Write(path=tempdir)
+            self._log_data = self.readLog(log_path)
+        self.assertEqual(len(self._log_data), 2)
+        error_event = self._log_data[1]
+        self.verifyCommonKeys(self._log_data[0], expected_event_name="version")
+        self.verifyCommonKeys(error_event, expected_event_name="error")
+        # Check for 'error' event specific fields.
+        self.assertIn("msg", error_event)
+        self.assertIn("fmt", error_event)
+        self.assertEqual(error_event["msg"], msg)
+        self.assertEqual(error_event["fmt"], fmt)
+    def test_write_with_filename(self):
+        """Test Write() with a path to a file exits with None."""
+        self.assertIsNone(self._event_log_module.Write(path="path/to/file"))
+    def test_write_with_git_config(self):
+        """Test Write() uses the git config path when 'git config' call
+        succeeds."""
+        with tempfile.TemporaryDirectory(prefix="event_log_tests") as tempdir:
+            with mock.patch.object(
+                self._event_log_module,
+                "_GetEventTargetPath",
+                return_value=tempdir,
+            ):
+                self.assertEqual(
+                    os.path.dirname(self._event_log_module.Write()), tempdir
+                )
+    def test_write_no_git_config(self):
+        """Test Write() with no git config variable present exits with None."""
+        with mock.patch.object(
+            self._event_log_module, "_GetEventTargetPath", return_value=None
+        ):
+            self.assertIsNone(self._event_log_module.Write())
+    def test_write_non_string(self):
+        """Test Write() with non-string type for |path| throws TypeError."""
+        with self.assertRaises(TypeError):
+            self._event_log_module.Write(path=1234)
+    def test_write_socket(self):
+        """Test Write() with Unix domain socket for |path| and validate received
+        traces."""
+        received_traces = []
+        with tempfile.TemporaryDirectory(
+            prefix="test_server_sockets"
+        ) as tempdir:
+            socket_path = os.path.join(tempdir, "server.sock")
+            server_ready = threading.Condition()
+            # Start "server" listening on Unix domain socket at socket_path.
+            try:
+                server_thread = threading.Thread(
+                    target=serverLoggingThread,
+                    args=(socket_path, server_ready, received_traces),
+                )
+                server_thread.start()
+                with server_ready:
+                    server_ready.wait(timeout=120)
+                self._event_log_module.StartEvent()
+                path = self._event_log_module.Write(
+                    path=f"af_unix:{socket_path}"
+                )
+            finally:
+                server_thread.join(timeout=5)
+        self.assertEqual(path, f"af_unix:stream:{socket_path}")
+        self.assertEqual(len(received_traces), 2)
+        version_event = json.loads(received_traces[0])
+        start_event = json.loads(received_traces[1])
+        self.verifyCommonKeys(version_event, expected_event_name="version")
+        self.verifyCommonKeys(start_event, expected_event_name="start")
+        # Check for 'start' event specific fields.
+        self.assertIn("argv", start_event)
+        self.assertIsInstance(start_event["argv"], list)