Merge 5c9660a into e4e8312

Pipfile

@@ -25,6 +25,7 @@ redis = "~=2.10"
 requests = ">=2.20"
 rethinkdb = "==2.3"
 pickle-mixin = "==1.0.2"
+pytest-lazy-fixture = "*"
 
 [dev-packages]
 

src/conftest.py

@@ -15,7 +15,7 @@
 CRATE_PORT = 4200
 
 POSTGRES_HOST = os.environ.get('POSTGRES_HOST', 'timescale')
-POSTGRES_HOST = 5432
+POSTGRES_PORT = 5432
 
 REDIS_HOST = os.environ.get('REDIS_HOST', 'redis')
 REDIS_PORT = 6379
@@ -184,6 +184,54 @@ def clean(self, fiware_service=None, **kwargs):
         yield trans
 
 
+@pytest.fixture()
+def timescale_translator():
+    from src.translators.timescale import PostgresTranslator, \
+        PostgresConnectionData
+
+    class Translator(PostgresTranslator):
+
+        def insert(self, entities,
+                   fiware_service=None, fiware_servicepath=None):
+            r = PostgresTranslator.insert(self, entities,
+                                          fiware_service, fiware_servicepath)
+            return r
+
+        def delete_entity(self, entity_id, entity_type=None,
+                          fiware_service=None, **kwargs):
+            r = PostgresTranslator.delete_entity(self, entity_id, entity_type,
+                                                 fiware_service=fiware_service,
+                                                 **kwargs)
+            return r
+
+        def delete_entities(self, entity_type=None, fiware_service=None,
+                            **kwargs):
+            r = PostgresTranslator.delete_entities(self, entity_type,
+                                                   fiware_service=fiware_service,
+                                                   **kwargs)
+            return r
+
+        def entity_types(self, fiware_service=None, **kwargs):
+            r = PostgresTranslator.query_entity_types(self, entity_type=None,
+                                                      fiware_service=fiware_service,
+                                                      **kwargs)
+            return r
+
+        def clean(self, fiware_service=None, **kwargs):
+            types = PostgresTranslator.query_entity_types(self,
+                                                          fiware_service=fiware_service,
+                                                          **kwargs)
+            if types:
+                for t in types:
+                    PostgresTranslator.drop_table(self, t,
+                                                  fiware_service=fiware_service,
+                                                  **kwargs)
+
+    with Translator(PostgresConnectionData(host=POSTGRES_HOST,
+                                           port=POSTGRES_PORT)) as trans:
+        yield trans
+
+
 @pytest.fixture
 def entity():
     entity = {
@@ -454,3 +502,71 @@ def traffic_flow_observed():
         }
     }
     return entity
+
+
+@pytest.fixture
+def ngsi_ld():
+    """
+    :return: dict
+        The NGSI LD model as received within an Orion notification.
+    """
+    entity = {
+        "id": "urn:ngsi-ld:Streetlight:streetlight:guadalajara:4567",
+        "type": "Streetlight",
+        "location": {
+            "type": "GeoProperty",
+            "value": {
+                "type": "Point",
+                "coordinates": [-3.164485591715449, 40.62785133667262]
+            }
+        },
+        "areaServed": {
+            "type": "Property",
+            "value": "Roundabouts city entrance"
+        },
+        "status": {
+            "type": "Property",
+            "value": "ok"
+        },
+        "refStreetlightGroup": {
+            "type": "Relationship",
+            "object": "urn:ngsi-ld:StreetlightGroup:streetlightgroup:G345"
+        },
+        "refStreetlightModel": {
+            "type": "Relationship",
+            "object": "urn:ngsi-ld:StreetlightModel:streetlightmodel:STEEL_Tubular_10m"
+        },
+        "circuit": {
+            "type": "Property",
+            "value": "C-456-A467"
+        },
+        "lanternHeight": {
+            "type": "Property",
+            "value": 10
+        },
+        "locationCategory": {
+            "type": "Property",
+            "value": "centralIsland"
+        },
+        "powerState": {
+            "type": "Property",
+            "value": "off"
+        },
+        "controllingMethod": {
+            "type": "Property",
+            "value": "individual"
+        },
+        "dateLastLampChange": {
+            "type": "Property",
+            "value": {
+                "@type": "DateTime",
+                "@value": "2016-07-08T08:02:21.753Z"
+            }
+        },
+        "@context": [
+            "https://schema.lab.fiware.org/ld/context",
+            "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context.jsonld"
+        ]
+    }
+
+    return entity

src/translators/sql_translator.py

@@ -437,7 +437,7 @@ def _insert_original_entities_in_failed_batch(
         self.cursor.executemany(stmt, rows)
 
     def _attr_is_structured(self, a):
-        if a['value'] is not None and isinstance(a['value'], dict):
+        if 'value' in a and a['value'] is not None and isinstance(a['value'], dict):
             self.logger.debug("attribute {} has 'value' attribute of type dict"
                               .format(a))
             return True

src/translators/tests/original_data_scenarios.py

@@ -11,10 +11,10 @@
     TYPE_PREFIX, TENANT_PREFIX
 from utils.jsondict import maybe_value
 
-
 ENTITY_TYPE = 'device'
 TranslatorFactory = Callable[[], Generator[SQLTranslator, Any, None]]
 
+
 #
 # NOTE. Each test scenario gets a (sort of) unique tenant so that we won't
 # have to clean up the DB after each test, which would slow down the whole
@@ -23,7 +23,7 @@
 
 
 def gen_tenant_id() -> str:
-    tid = random.randint(1, 2**32)
+    tid = random.randint(1, 2 ** 32)
     return f"tenant{tid}"
 
 
@@ -55,9 +55,9 @@ def assert_saved_original(actual_row, original_entity,
 
 def assert_inserted_entity(actual_row, original_entity):
     assert actual_row['a_number'] == \
-        maybe_value(original_entity, 'a_number', 'value')
+           maybe_value(original_entity, 'a_number', 'value')
     assert actual_row['an_attr'] == \
-        maybe_value(original_entity, 'an_attr', 'value')
+           maybe_value(original_entity, 'an_attr', 'value')
     assert actual_row[ORIGINAL_ENTITY_COL] is None
 
 
@@ -82,6 +82,9 @@ def __init__(self, translator: TranslatorFactory, cursor,
         self.cursor = cursor
         self.delay_query_by = delay_query_by
 
+    def get_translator(self):
+        return self.translator
+
     def insert_entities(self, tenant: str, entities: List[dict]):
         with self.translator() as t:
             t.insert(entities, fiware_service=tenant)

src/translators/tests/test_crate_aggregation.py → src/translators/tests/test_aggregation.py

@@ -1,8 +1,23 @@
-from conftest import crate_translator as translator
+# To test a single translator use the -k parameter followed by either
+# timescale or crate.
+# See https://docs.pytest.org/en/stable/example/parametrize.html
+
+from conftest import crate_translator, timescale_translator
 from utils.common import create_random_entities, TIME_INDEX_NAME, add_attr
 import datetime
 
+from src.utils.common import create_random_entities
+
+import pytest
+
+
+translators = [
+    pytest.lazy_fixture('crate_translator'),
+    pytest.lazy_fixture('timescale_translator')
+]
+
 
+@pytest.mark.parametrize("translator", translators, ids=["crate", "timescale"])
 def test_aggr_per_second(translator):
     entities = create_random_entities(num_ids_per_type=2, num_updates=17)
     assert len(entities) == 34
@@ -47,4 +62,4 @@ def test_aggr_per_second(translator):
             'values': [5, 15, 25, 32],
         }
     }
-    translator.clean()
+    translator.clean()