Merge 5fe26b4 into cacc0f6

.gitignore

@@ -6,7 +6,7 @@ build
 MANIFEST
 mongo-connector.log
 oplog.timestamp
-mappings.json
+/mappings.json
 mappings_bak.json
 run-db.sh
 .*.swp
@@ -20,3 +20,4 @@ venv
 mongo_connector_postgresql.egg-info
 .coverage
 htmlcov
+server.log

.travis.yml

@@ -3,8 +3,19 @@ python:
   - "2.7"
   - "3.4"
   - "3.5"
+services:
+    - postgresql
+addons:
+  apt:
+    sources:
+      - mongodb-upstart
+      - mongodb-3.4-precise
+    packages:
+      - mongodb-org
 before_install:
   - pip install --upgrade pip setuptools
+  - pip install mock>=2.0.0
+  - pip install -r requirements.test.txt
 install:
   - pip install coveralls
   - python setup.py install

features/.gitignore

@@ -0,0 +1,3 @@
+config.json
+mongo-connector.log
+oplog.timestamp

features/__init__.py

@@ -0,0 +1,205 @@
+# -*- coding: utf-8 -*-
+
+from contextlib import contextmanager
+from aloe import world, step, around
+from testing.postgresql import PostgresqlFactory
+
+from psycopg2 import connect as pg_connect
+from psycopg2.extras import RealDictCursor
+from pymongo import MongoClient
+
+from future.moves.urllib.parse import urlparse, parse_qs
+from time import sleep
+import subprocess
+import requests
+import json
+import os
+
+
+@around.all
+@contextmanager
+def handle_pg_factory():
+    world.DEVNULL = open('/dev/null', 'w')
+    world.Postgresql = PostgresqlFactory(cache_initialized_db=True)
+    world.Postgresql.clear_cache()
+    yield
+    world.Postgresql.clear_cache()
+    world.DEVNULL.close()
+
+
+@around.each_example
+@contextmanager
+def handle_databases(scenario, outline, steps):
+    world.initial_dir = os.getcwd()
+    world.result = {}
+    world.envvars = {}
+
+    info = requests.post(
+        'http://localhost:8889/replica_sets',
+        data=json.dumps({
+            'id': 'rs0',
+            'members': [{}]
+        })
+    )
+
+    assert info.status_code == 200, \
+        "Impossible to create replicat set: {}".format(info.status_code)
+
+    world.mongo_server = info.json().get('server_id')
+    world.mongo_uri = info.json().get('mongodb_uri')
+    world.mongo_conn = MongoClient(world.mongo_uri, w=1, j=True)
+
+    world.pg_server = world.Postgresql()
+    world.pg_uri = world.pg_server.url()
+    world.pg_conn = pg_connect(world.pg_uri)
+
+    try:
+        yield
+
+    finally:
+        world.mongo_conn.close()
+        world.pg_conn.close()
+
+        if hasattr(world, 'mongo_connector'):
+            world.mongo_connector.terminate()
+            world.mongo_connector.wait()
+
+        ret = requests.delete(
+            'http://localhost:8889/replica_sets/rs0'
+        )
+        assert ret.status_code == 204, \
+            "Impossible to delete replica set: {}".format(ret.status_code)
+
+        del world.mongo_server
+        del world.mongo_uri
+
+        world.pg_server.stop()
+
+        os.chdir(world.initial_dir)
+        del world.result
+        del world.envvars
+
+
+@step('I have the environment "([^"]*)"')
+def initialize_environ(self, environ):
+    os.chdir(os.path.join('features', 'envs', environ))
+
+    with open('envvars.json') as f:
+        world.envvars = json.load(f)
+
+    config = {
+        'mainAddress': world.mongo_uri,
+        'docManagers': [
+            {
+                'docManager': 'postgresql_manager',
+                'targetURL': world.pg_uri,
+                'args': {
+                    'mongoUrl': world.mongo_uri
+                }
+            }
+        ]
+    }
+
+    with open('config.json', 'w') as f:
+        json.dump(config, f)
+
+    parts = urlparse(world.mongo_uri)
+    mongo_host = parts.hostname
+    mongo_port = parts.port
+    mongorestore = subprocess.Popen(
+        'mongorestore -h rs0/{} --port {} -d {} dump'.format(
+            mongo_host,
+            mongo_port,
+            world.envvars['DBNAME']
+        ),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        shell=True
+    )
+    retcode = mongorestore.wait()
+    assert retcode == 0, \
+        "Impossible to insert dump in MongoDB:\nOUT: {}\nERR: {}".format(
+            *mongorestore.communicate()
+        )
+
+    world.retries = world.envvars.get('RETRIES', 3)
+
+
+@step('I run mongo-connector')
+def run_mongo_connector(self):
+    trailing = ['mongo-connector.log', 'oplog.timestamp']
+    for filename in trailing:
+        if os.path.exists(filename):
+            os.remove(filename)
+
+    world.mongo_connector = subprocess.Popen(
+        "mongo-connector -c {0}/config.json".format(os.getcwd()),
+        shell=True,
+        stdout=world.DEVNULL,
+        stderr=world.DEVNULL
+    )
+
+
+@step('I wait (\d+) seconds for the replication to be done')
+def wait_for_replication(self, seconds):
+    sleep(int(seconds))
+    world.wait = int(seconds)
+
+
+@step('I delete documents from the collection')
+def delete_collection(self):
+    db = world.mongo_conn[world.envvars['DBNAME']]
+    collection = db[world.envvars['COLLECTION']]
+    spec = world.envvars.get('SPEC', {})
+    collection.delete_many(spec)
+
+
+@step('I update the collection')
+def update_collection(self):
+    db = world.mongo_conn[world.envvars['DBNAME']]
+    collection = db[world.envvars['COLLECTION']]
+    spec = world.envvars.get('SPEC', {})
+    update = world.envvars['UPDATE']
+    collection.update_many(spec, update)
+
+
+@step('I run the SQL queries')
+def run_sql_queries(self):
+    for query in world.envvars['QUERIES']:
+        with world.pg_conn.cursor(
+            cursor_factory=RealDictCursor
+        ) as cursor:
+            cursor.execute(query['sql'])
+            world.result[query['name']] = cursor.fetchall()
+
+
+@step('the SQL queries should return the appropriate results')
+def check_queries(self):
+    success = True
+    messages = []
+
+    while world.retries > 0:
+        messages = []
+
+        for query in world.envvars['QUERIES']:
+            qname = query['name']
+            result = world.result[qname]
+            expected = query['expected']
+
+            if result != expected:
+                msg = 'Query {} did not return appropriate result: {}'.format(
+                    qname,
+                    result
+                )
+                success = False
+                messages.append(msg)
+
+        if success:
+            break
+
+        world.retries -= 1
+        sleep(world.wait)
+        run_sql_queries(self)
+
+    if not success:
+        raise AssertionError('\n'.join(messages))

features/envs/replicate_deletions/dump/col_test.metadata.json

@@ -0,0 +1 @@
+{"options":{},"indexes":[{"v":1,"key":{"_id":1},"name":"_id_","ns":"db_test.col_test"}]}

features/envs/replicate_deletions/envvars.json

@@ -0,0 +1,16 @@
+{
+    "DBNAME": "db_test",
+    "COLLECTION": "col_test",
+    "SPEC": {
+        "a": {"$eq": 5}
+    },
+    "QUERIES": [
+        {
+            "name": "count_entries",
+            "sql": "SELECT COUNT(*) AS c FROM col_test",
+            "expected": [
+                {"c": 1}
+            ]
+        }
+    ]
+}

features/envs/replicate_deletions/mappings.json

@@ -0,0 +1,13 @@
+{
+    "db_test": {
+        "col_test": {
+            "pk": "_id",
+            "_id": {
+                "type": "INT"
+            },
+            "a": {
+                "type": "INT"
+            }
+        }
+    }
+}

features/envs/replicate_inserts/dump/col_test.metadata.json

@@ -0,0 +1 @@
+{"options":{},"indexes":[{"v":1,"key":{"_id":1},"name":"_id_","ns":"db_test.col_test"}]}

features/envs/replicate_inserts/envvars.json

@@ -0,0 +1,13 @@
+{
+    "DBNAME": "db_test",
+    "COLLECTION": "col_test",
+    "QUERIES": [
+        {
+            "name": "count_entries",
+            "sql": "SELECT COUNT(*) AS c FROM col_test",
+            "expected": [
+                {"c": 2}
+            ]
+        }
+    ]
+}

features/envs/replicate_inserts/mappings.json

@@ -0,0 +1,13 @@
+{
+    "db_test": {
+        "col_test": {
+            "pk": "_id",
+            "_id": {
+                "type": "INT"
+            },
+            "a": {
+                "type": "INT"
+            }
+        }
+    }
+}

features/envs/replicate_list_of_documents/dump/col_test_array.metadata.json

@@ -0,0 +1 @@
+{"options":{},"indexes":[{"v":2,"key":{"_id":1},"name":"_id_","ns":"db_test_array.col_test_array"}]}

features/envs/replicate_list_of_documents/envvars.json

@@ -0,0 +1,20 @@
+{
+    "DBNAME": "db_test_array",
+    "COLLECTION": "col_test_array",
+    "QUERIES": [
+        {
+            "name": "count_entries",
+            "sql": "SELECT COUNT(*) AS c FROM col_test_array",
+            "expected": [
+                {"c": 5}
+            ]
+        },
+        {
+            "name": "count_nested_entries",
+            "sql": "SELECT COUNT(*) AS c FROM col_array WHERE id_test = 1",
+            "expected": [
+                {"c": 5}
+            ]
+        }
+    ]
+}

features/envs/replicate_list_of_documents/mappings.json

@@ -0,0 +1,24 @@
+{
+    "db_test_array": {
+        "col_test_array": {
+            "pk": "_id",
+            "_id": {
+                "type": "INT"
+            },
+            "a": {
+                "type": "_ARRAY",
+                "dest": "col_array",
+                "fk": "id_test"
+            }
+        },
+        "col_array": {
+            "pk": "_id",
+            "id_test": {
+                "type": "INT"
+            },
+            "b": {
+                "type": "INT"
+            }
+        }
+    }
+}

features/envs/replicate_list_of_documents_nested/dump/col_nested.metadata.json

@@ -0,0 +1 @@
+{"options":{},"indexes":[{"v":1,"key":{"_id":1},"name":"_id_","ns":"db_test_nested_array.col_nested"}]}

features/envs/replicate_list_of_documents_nested/envvars.json

@@ -0,0 +1,27 @@
+{
+    "DBNAME": "db_test_nested_array",
+    "COLLECTION": "col_nested",
+    "QUERIES": [
+        {
+            "name": "count_entries",
+            "sql": "SELECT COUNT(*) AS c FROM col_nested",
+            "expected": [
+                {"c": 1}
+            ]
+        },
+        {
+            "name": "count_entries_lvl1",
+            "sql": "SELECT COUNT(*) AS c FROM col_array_lvl1 WHERE id_nested = 1",
+            "expected": [
+                {"c": 3}
+            ]
+        },
+        {
+            "name": "count_entries_lvl2",
+            "sql": "SELECT COUNT(*) AS c FROM col_array_lvl2 WHERE id_lvl1 = 1",
+            "expected": [
+                {"c": 5}
+            ]
+        }
+    ]
+}