refactor: command line

rerwatcher/__main__.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # coding: utf-8
+import argparse
 import os
-import sys
 
 from loguru import logger
 
@@ -14,20 +14,21 @@
 logger.add(log_success, rotation='00:00', retention='2 days', level='DEBUG')
 logger.add(log_error, rotation='00:00', retention='2 days', level='ERROR')
 
+
+def parse_operation():
+    choices = ['start', 'stop', 'status']
+    parser = argparse.ArgumentParser(description="RERWatcher")
+    parser.add_argument('operation', type=str, choices=choices)
+    args = parser.parse_args()
+    return args.operation
+
+
 daemon = RerWatcher(
     pidfile='/tmp/rerwatcher.pid',
     stdout=log_success,
     stderr=log_error
 )
 
-if len(sys.argv) != 2:
-    print(f"Usage: {sys.argv[0]} [start|stop].", file=sys.stderr)
-    raise SystemExit(1)
-
-if 'start' == sys.argv[1]:
-    daemon.start()
-elif 'stop' == sys.argv[1]:
-    daemon.stop()
-else:
-    print(f"Unknown command {sys.argv[1]!r}.", file=sys.stderr)
-    raise SystemExit(1)
+operation = parse_operation()
+operation = getattr(daemon, operation)
+operation()

rerwatcher/daemon.py

@@ -59,7 +59,7 @@ def fork_os():
         except OSError:
             raise RuntimeError("Fork failed.")
 
-    def _is_running(self):
+    def _is_running(self) -> bool:
         return os.path.exists(self.pidfile)
 
     def start(self):
@@ -75,6 +75,13 @@ def stop(self):
             pid = int(f.read())
             os.kill(pid, signal.SIGTERM)
 
+    def status(self):
+        status = [
+            f"{self.app_name} is not running.",
+            f"{self.app_name} is running.",
+        ]
+        print(status[self._is_running()])
+
     @abstractmethod
     def run(self):
         raise NotImplementedError

tests/test_daemon.py

@@ -64,3 +64,20 @@ def test_daemon_do_not_stop_if_not_running(mocker):
 
     with pytest.raises(SystemExit):
         daemon_app.stop()
+
+
+@pytest.mark.parametrize('is_running, expected', [
+    (True, 'TestApp is running.\n'),
+    (False, 'TestApp is not running.\n'),
+])
+def test_daemon_status(mocker, capsys, is_running, expected):
+    mocker.patch(
+        'rerwatcher.daemon.Daemon._is_running',
+        return_value=is_running
+    )
+    daemon_app = DaemonApp()
+
+    daemon_app.status()
+
+    captured = capsys.readouterr()
+    assert expected == captured.out