Basics
authorJack Miller <jack@codezen.org>
Wed, 9 Jun 2010 02:56:17 +0000 (21:56 -0500)
committerJack Miller <jack@codezen.org>
Wed, 9 Jun 2010 02:56:17 +0000 (21:56 -0500)
Signed-off-by: Jack Miller <jack@codezen.org>
canto_curses/__init__.py [new file with mode: 0644]
canto_curses/main.py [new file with mode: 0755]
test.py [new file with mode: 0755]
tests/__init__.py [new file with mode: 0644]
tests/main.py [new file with mode: 0644]

diff --git a/canto_curses/__init__.py b/canto_curses/__init__.py
new file mode 100644 (file)
index 0000000..e69de29
diff --git a/canto_curses/main.py b/canto_curses/main.py
new file mode 100755 (executable)
index 0000000..eba42f8
--- /dev/null
@@ -0,0 +1,108 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#Canto-curses - ncurses RSS reader
+#   Copyright (C) 2010 Jack Miller <jack@codezen.org>
+#
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License version 2 as 
+#   published by the Free Software Foundation.
+
+from canto.client import CantoClient
+from canto.encoding import decoder
+
+import logging
+
+logging.basicConfig(
+        filemode = "w",
+        format = "%(asctime)s : %(name)s -> %(message)s",
+        datefmt = "%H:%M:%S",
+        level = logging.DEBUG
+)
+
+log = logging.getLogger("CANTO-CURSES")
+
+import traceback
+import getopt
+import errno
+import fcntl
+import time
+import sys
+import os
+
+class CantoCurses(CantoClient):
+
+    # Init separate from instantiation for test purposes.
+    def __init__(self):
+        pass
+
+    def init(self, args=None):
+        if self.args(args):
+            sys.exit(-1)
+
+        self.start_daemon()
+
+        # The daemon is backed, init our base class,
+        # start trying to connect to the daemon.
+
+        try:
+            CantoClient.__init__(self, self.socket_path)
+        except Exception, e:
+            log.error("Error: %s" % e)
+            sys.exit(-1)
+
+    def run(self):
+        while True:
+            time.sleep(0.01)
+
+    def args(self, args):
+        if not args:
+            args = sys.argv[1:]
+
+        try:
+            optlist = getopt.getopt(args, 'D:', ["dir="])[0]
+        except getopt.GetoptError, e:
+            log.error("Error: %s" % e.msg)
+
+        self.conf_dir = os.path.expanduser(u"~/.canto-ng/")
+
+        for opt, arg in optlist:
+            if opt in [ "-D", "--dir"]:
+                self.conf_dir = os.path.expanduser(decoder(arg))
+                self.conf_dir = os.path.realpath(self.conf_dir)
+
+        self.socket_path = self.conf_dir + "/.canto_socket"
+
+        return 0
+
+    def start_daemon(self):
+        pid = os.fork()
+        if not pid:
+            os.execve("/bin/sh",
+                     ["/bin/sh", "-c", "canto-daemon -D " + self.conf_dir],
+                     os.environ)
+
+            # Should never get here, but just in case.
+            sys.exit(-1)
+
+        while not os.path.exists(self.socket_path):
+            time.sleep(0.1)
+
+        return pid
+
+    def start(self, args=None):
+        try:
+            self.init(args)
+            self.run()
+        except KeyboardInterrupt:
+            pass
+
+        except Exception, e:
+            tb = traceback.format_exc(e)
+            log.error("Exiting on exception:")
+            log.error("\n" + "".join(tb))
+
+        sys.exit(0)
+
+if __name__ == "__main__":
+    c = CantoCurses()
+    c.start()
diff --git a/test.py b/test.py
new file mode 100755 (executable)
index 0000000..9d7e7c4
--- /dev/null
+++ b/test.py
@@ -0,0 +1,51 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+#Canto - RSS reader backend
+#   Copyright (C) 2010 Jack Miller <jack@codezen.org>
+#
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License version 2 as 
+#   published by the Free Software Foundation.
+
+import tests.main
+
+import logging
+
+logging.basicConfig(
+    filemode = "w",
+    format = "%(asctime)s : %(name)s -> %(message)s",
+    datefmt = "%H:%M:%S",
+    level = logging.DEBUG
+)
+
+import unittest
+import sys
+
+all_modules = {
+        "main" : tests.main.Tests }
+
+all_tests = {
+        "test_args" : tests.main.Tests,
+        "test_start_daemon" : tests.main.Tests
+}
+
+if __name__ == "__main__":
+    t = []
+    if len(sys.argv) == 1:
+        for key in all_tests:
+            t.append(all_tests[key](key))
+    else:
+        for arg in sys.argv[1:]:
+            if arg in all_tests:
+                t.append(all_tests[arg](arg))
+            elif arg in all_modules:
+                for k in all_tests:
+                    if all_tests[k] == all_modules[arg]:
+                        t.append(all_tests[k](k))
+            else:
+                print "Unknown arg: %s" % arg
+
+    suite = unittest.TestSuite()
+    suite.addTests(t)
+    unittest.TextTestRunner(verbosity=2).run(suite)
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644 (file)
index 0000000..e69de29
diff --git a/tests/main.py b/tests/main.py
new file mode 100644 (file)
index 0000000..52173db
--- /dev/null
@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+#Canto-curses - ncurses RSS reader
+#   Copyright (C) 2010 Jack Miller <jack@codezen.org>
+#
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License version 2 as 
+#   published by the Free Software Foundation.
+
+from canto_curses.main import CantoCurses
+from canto.client import CantoClient
+
+import unittest
+import signal
+import os
+
+class Tests(unittest.TestCase):
+
+    def test_args(self):
+        b = CantoCurses()
+
+        # Test no arguments
+        b.args([])
+        self.assertEqual(b.conf_dir, os.getenv("HOME") + "/.canto-ng/")
+        self.assertEqual(type(b.conf_dir), unicode)
+
+        # Test -D initial directory setup
+        b.args(["-D", "/some/path/somewhere"])
+        self.assertEqual(b.conf_dir, "/some/path/somewhere")
+        self.assertEqual(type(b.conf_dir), unicode)
+
+    def test_start_daemon(self):
+        b = CantoCurses()
+        d = os.getenv("PWD") + "/tests/basic_dir"
+        s = d + "/.canto_socket"
+
+        if os.path.exists(s):
+            os.remove(s)
+
+        b.args(["-D", d])
+        pid = b.start_daemon()
+
+        # Actually connect to the socket. If
+        # there isn't a daemon running on it,
+        # ECONNREFUSED exception will fail the test.
+        CantoClient.__init__(b, s)
+
+        # Don't leave it hanging around for no reason.
+        # os.kill(pid, signal.SIGKILL)