not_complicated_at_all.py

# coding: spec

from convex_scheduler.db import DatabaseConnection
from convex_scheduler import hooks

from photons_app.test_helpers import TestCase

from contextlib import contextmanager
from datetime import datetime
import uuid
import mock

describe TestCase, "hooks":

    @contextmanager
    def a_database(self):
        db = DatabaseConnection('postgresql:///test_triggers_platform')
        db.create_tables()
        @contextmanager
        def new_session():
            with db.new_session() as d:
                yield d
        yield new_session
        db.close()

    describe "hook_to_triggers":
        it "returns triggers":
            token = str(uuid.uuid1())
            source = "lifx"
            selector = str(uuid.uuid1())
            time_spec = str(uuid.uuid1())
            account_id = str(uuid.uuid1())
            event_type = str(uuid.uuid1())
            current_value = {}

            timepiece_parser = mock.Mock(name="timepiece_parser")

            with self.a_database() as session_maker:
                with session_maker() as db:
                    db.add(db.queries.create_definition(
                          account=account_id, token=token, selector=selector, action="state"
                        , options={"power": "on"}, trigger_time_spec=time_spec
                        , submission_time = datetime.utcnow()
                        , conditions = [db.queries.create_condition(source=source, event_type=event_type, account_id=account_id, match_expression="True")]
                        )
                    )
                    db.commit()

                with session_maker() as db:
                    triggers = hooks.hook_to_triggers(source, account_id, event_type, current_value, timepiece_parser, db)
                    self.assertIs(len(triggers), 1)
                    conditions = [
                        { "account_id": account_id, "current_value": None
                        , "event_type": event_type, "last_updated": None
                        , "last_value": None, "match_expression": "True", "source": "lifx"
                        }
                    ]

                    self.assertEqual(triggers[0].as_dict()
                        , { "action": "state"
                          , "at": None
                          , "conditions": conditions
                          , "definition_id": 1
                          , "filtered": False
                          , "options": {"power": "on"}
                          , "selector": selector
                          , "token": token
                          }
                        )