Merge pull request #1 from astrohr/patrik_develop

pi ar for pi ar

setup.py

@@ -19,6 +19,7 @@
         'astropy>=3.1.1,<3.2',
         'psycopg2>=2.7,<2.8',
         'numpy>=1.11.0,<1.16',
+        'python-dotenv>=0.10.1,<0.11',
     ],
     extras_require={'dev': ['flake8', 'black', 'coverage']},
 )

void/common.py

@@ -4,6 +4,7 @@
 """
 
 import logging
+import psycopg2
 
 import docopt
 
@@ -12,6 +13,9 @@
 )
 
 
+log = logging.getLogger(__name__)
+
+
 def configure_log(verbosity):
     levels = {
         '0': logging.CRITICAL,
@@ -23,3 +27,46 @@ def configure_log(verbosity):
     if verbosity not in levels.keys():
         raise docopt.DocoptExit('--verbosity not one of 0, 1, 2, 3, 4')
     logging.basicConfig(level=levels[verbosity], format=LOG_FORMAT)
+
+
+class DataBase:
+    def __init__(self, user, passwd, db_name, host, port):
+        self.conn = psycopg2.connect(
+            database=db_name, user=user, password=passwd, host=host, port=port
+        )
+        self.cursor = self.conn.cursor()
+        log.debug('connection established')
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+        return False
+
+    def exec(self, exe_str, *attr):
+        self.cursor.execute(exe_str, attr)
+        self.conn.commit()
+
+    def close(self):
+        self.cursor.close()
+
+    @classmethod
+    def get_pg_db(cls, settings):
+        return cls(
+            user=settings.POSTGRES_USER,
+            passwd=settings.POSTGRES_PASSWORD,
+            db_name='postgres',
+            host='localhost',
+            port='5433',
+        )
+
+    @classmethod
+    def get_void_db(cls, settings):
+        return cls(
+            user=settings.POSTGRES_USER,
+            passwd=settings.POSTGRES_PASSWORD,
+            db_name=settings.POSTGRES_DB,
+            host='localhost',
+            port='5433',
+        )

void/settings.py

@@ -0,0 +1,26 @@
+import os
+
+from dotenv import load_dotenv
+
+
+class Settings:
+    _loaded = False
+
+    defaults = {
+        'POSTGRES_USER': 'void',
+        'POSTGRES_PASSWORD': 'void',
+        'POSTGRES_DB': 'void',
+    }
+
+    def load(self, env_file='/Users/fran/.voidrc'):
+        if not self._loaded:
+            load_dotenv(env_file, verbose=True)
+            self._loaded = True
+
+    def __getattr__(self, item):
+        if not self._loaded:
+            raise RuntimeError('Settings not yet loaded.')
+        return os.getenv(item, self.defaults.get(item))
+
+
+settings = Settings()

void/tests/test_writer.py

@@ -0,0 +1,137 @@
+import os
+import unittest
+from unittest import mock
+
+import psycopg2
+
+from void import common
+from void.settings import settings
+from void.writer import Writer, main
+
+
+class WriterTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        env_path = os.path.join(os.path.dirname(__file__), 'tests.env')
+        settings.load(env_path)
+
+        with common.DataBase.get_pg_db(settings) as db:
+            db.conn.set_isolation_level(0)
+            db.exec(f'DROP DATABASE IF EXISTS {settings.POSTGRES_DB};')
+            db.exec(f'CREATE DATABASE {settings.POSTGRES_DB};')
+            db.conn.set_isolation_level(1)
+
+        with common.DataBase.get_void_db(settings) as db:
+            db.exec('DROP TABLE IF EXISTS observations;')
+            db.exec('DROP EXTENSION IF EXISTS postgis CASCADE;')
+
+    @classmethod
+    def tearDownClass(cls):
+        with common.DataBase.get_pg_db(settings) as db:
+            db.conn.set_isolation_level(0)
+            db.exec(f'DROP DATABASE {settings.POSTGRES_DB};')
+            db.conn.set_isolation_level(1)
+
+    @mock.patch('void.writer.Writer.create_table')
+    def test_init(self, p_create_table):
+        writer = Writer()
+        p_create_table.assert_called_once_with()
+        self.assertIsInstance(writer.db, common.DataBase)
+
+    def test_create_table(self):
+        with mock.patch.object(Writer, 'create_table'):
+            writer = Writer()
+        with common.DataBase.get_void_db(settings) as db:
+            with self.assertRaises(psycopg2.ProgrammingError):
+                db.exec('SELECT * FROM observations LIMIT 2')
+
+        writer.create_table()
+        writer.close()
+
+        with common.DataBase.get_void_db(settings) as db:
+            results = db.exec('SELECT * FROM observations LIMIT 2')
+        self.assertIsNone(results)
+
+    def test_decode_data(self):
+        writer = Writer()
+        input_str = '{"foo": "bar"}'
+        expected = {'foo': 'bar'}
+        actual = writer.decode_data(input_str)
+        self.assertEqual(expected, actual)
+
+    def test_poly_to_linestr(self):
+        writer = Writer()
+        poly = [[1, 2], [3, 4], [5, 6]]
+        date_tstamp = 30
+        expected = "1 2 30,3 4 30,5 6 30,1 2 30"
+        actual = writer.poly_to_linestr(date_tstamp, poly)
+        self.assertEqual(expected, actual)
+
+    def test_insert_data(self):
+        writer = Writer()
+        with common.DataBase.get_void_db(settings) as db:
+            db.exec('TRUNCATE TABLE observations;')
+        data_str = (
+            '{"path": '
+            '"/home/patrik/Workspace/void/void/tests/data/test2_flagged.fit", '
+            '"date_obs": "2018-12-26T18:41:49.300", '
+            '"exposure": 60.0, "observer": "", '
+            '"polygon": [[23.120172630110357, 30.275351191536046], '
+            '[23.080000853796616, 31.00642256430581], '
+            '[23.808858702289644, 31.046472709263952], '
+            '[23.849030478603385, 30.315401336494187]]}'
+        )
+        writer.insert_data(data_str)
+        with common.DataBase.get_void_db(settings) as db:
+            db.exec('SELECT * FROM observations')
+            records = db.cursor.fetchall()
+            self.assertIsNotNone(records)
+
+
+class MainTests(unittest.TestCase):
+
+    @mock.patch('void.writer.Writer')
+    @mock.patch('void.writer.log')
+    @mock.patch('void.writer.common')
+    @mock.patch('void.writer.sys')
+    @mock.patch('docopt.sys')
+    def test_cli_args(self, p_docopt_sys, p_sys, p_common, *_):
+        p_docopt_sys.argv = ['script.py', '-V', '3']
+        p_sys.stdin = []
+        main()
+        p_common.configure_log.assert_called_once_with('3')
+
+    @mock.patch('void.writer.Writer')
+    @mock.patch('void.writer.common')
+    @mock.patch('void.writer.docopt.docopt')
+    @mock.patch('void.writer.log')
+    @mock.patch('void.writer.sys')
+    def test_sigint(self, p_sys, p_log, *_):
+        mock_line = mock.MagicMock()
+        mock_line.strip.side_effect = KeyboardInterrupt
+        p_sys.stdin = [mock_line]
+        main()
+        self.assertIn(mock.call('SIGINT'), p_log.debug.call_args_list)
+
+    @mock.patch('void.writer.common')
+    @mock.patch('void.writer.docopt.docopt')
+    @mock.patch('void.writer.Writer')
+    @mock.patch('void.writer.sys')
+    def test_empty_line(self, p_sys, p_writer_cls, *_):
+        p_sys.stdin = ['line 1', '', 'line 3']
+        main()
+        writer = p_writer_cls.return_value
+        expected = [mock.call('line 1'), mock.call('line 3')]
+        self.assertEqual(expected, writer.insert_data.call_args_list)
+
+    @mock.patch('void.writer.common')
+    @mock.patch('void.writer.docopt.docopt')
+    @mock.patch('void.writer.Writer')
+    @mock.patch('void.writer.log')
+    @mock.patch('void.writer.sys')
+    def test_catch_exc(self, p_sys, p_log, p_writer_cls, *_):
+        p_sys.stdin = ['line 1']
+        writer = p_writer_cls.return_value
+        writer.insert_data.side_effect = ValueError('foo')
+        main()
+        p_log.warning.assert_called_once_with('foo', exc_info=True)

void/tests/tests.env

@@ -0,0 +1,3 @@
+POSTGRES_USER="void"
+POSTGRES_PASSWORD="void"
+POSTGRES_DB="void_test"

void/writer.py

@@ -21,49 +21,35 @@
 import sys
 
 import docopt
-import psycopg2
 from astropy.time import Time
 
 from void import common
+from void.settings import settings
+from void.common import DataBase
 
 log = logging.getLogger(__name__)
 
 
 class Writer:
     def __init__(self):
-        self.db_name = 'void'
-        self.user = 'void'
-        self.passwd = 'void'
-        self.host = 'localhost'
-        self.port = '5433'
-
-        self.conn = psycopg2.connect(
-            database=self.db_name,
-            user=self.user,
-            password=self.passwd,
-            host=self.host,
-            port=self.port,
-        )
-
-        self.cursor = self.conn.cursor()
-        log.debug('connection established')
-
+        settings.load()
+        self.db = DataBase.get_void_db(settings)
         self.create_table()
-        log.debug('table created')
 
     def create_table(self):
         """
         Creates a table if one does not already exist in VOID.
         """
-        self.cursor.execute('CREATE EXTENSION IF NOT EXISTS postgis;')
-        self.cursor.execute(
+        self.db.exec('CREATE EXTENSION IF NOT EXISTS postgis;')
+        self.db.exec(
             'CREATE TABLE IF NOT EXISTS observations '
             '(id SERIAL, '
             'path VARCHAR (500) NOT NULL, '
             'exp FLOAT NOT NULL, '
             'observer VARCHAR (100), '
             'poly GEOMETRY(POLYGONZ) NOT NULL);'
         )
+        log.debug('table created')
 
     @staticmethod
     def decode_data(data_str):
@@ -86,18 +72,16 @@ def insert_data(self, data_str):
         poly_str = "LINESTRING({:s})".format(
             self.poly_to_linestr(date_tstamp, poly)
         )
-
         exe_str = """
             INSERT INTO observations (path, exp, observer, poly)
             VALUES (%s, %s, %s,
             ST_MakePolygon(ST_GeomFromText(%s)));
         """
         log.debug(f'exe_str: {exe_str}')
-        self.cursor.execute(exe_str, (path, str(exp), observer, poly_str))
+        self.db.exec(exe_str, path, str(exp), observer, poly_str)
 
     def close(self):
-        self.conn.commit()
-        self.cursor.close()
+        self.db.close()
 
 
 def main():
@@ -125,5 +109,5 @@ def main():
     writer.close()
 
 
-if __name__ == '__main__':
+if __name__ == '__main__':  # pragma no cover
     main()