Tests: Added test for config

Author: erezsh

data_diff/__main__.py

@@ -12,7 +12,7 @@
 )
 from .databases.connect import connect
 from .parse_time import parse_time_before_now, UNITS_STR, ParseError
-from .config import apply_config
+from .config import apply_config_from_file
 
 import rich
 import click
@@ -63,13 +63,22 @@
     "A higher number will increase performance, but take more capacity from your database. "
     "'serial' guarantees a single-threaded execution of the algorithm (useful for debugging).",
 )
-@click.option("--conf", default=None, help="Path to a configuration.toml file, to provide a default configuration, and a list of possible runs.")
-@click.option("--run", default=None, help="Name of run-configuration to run. If used, CLI arguments for database and table must be omitted.")
+@click.option(
+    "--conf",
+    default=None,
+    help="Path to a configuration.toml file, to provide a default configuration, and a list of possible runs.",
+)
+@click.option(
+    "--run",
+    default=None,
+    help="Name of run-configuration to run. If used, CLI arguments for database and table must be omitted.",
+)
 def main(conf, run, **kw):
     if conf:
-        kw = apply_config(conf, run, kw)
+        kw = apply_config_from_file(conf, run, kw)
     return _main(**kw)
 
+
 def _main(
     database1,
     table1,
@@ -109,7 +118,7 @@ def _main(
         logging.error("Cannot specify a limit when using the -s/--stats switch")
         return
 
-    key_column = key_column or 'id'
+    key_column = key_column or "id"
     if bisection_factor is None:
         bisection_factor = DEFAULT_BISECTION_FACTOR
     if bisection_threshold is None:

data_diff/config.py

@@ -1,67 +1,73 @@
+from typing import Any, Dict
 import toml
-import logging
+
 
 class ConfigParseError(Exception):
     pass
 
 
 def is_uri(s: str) -> bool:
-    return '://' in s
-
+    return "://" in s
 
 
-def _load_config(path):
-    with open(path) as f:
-        return toml.load(f)
-
-def apply_config(path, run_name, kw):
+def _apply_config(config: Dict[str, Any], run_name: str, kw: Dict[str, Any]):
     # Load config
-    config = _load_config(path)
-    databases = config.pop('database', {})
-    runs = config.pop('run', {})
+    databases = config.pop("database", {})
+    runs = config.pop("run", {})
     if config:
         raise ConfigParseError(f"Unknown option(s): {config}")
 
     # Init run_args
-    run_args = runs.get('default') or {}
+    run_args = runs.get("default") or {}
     if run_name:
         if run_name not in runs:
-            raise ConfigParseError(f"Cannot find run '{run_name}' in configuration '{path}'.")
+            raise ConfigParseError(f"Cannot find run '{run_name}' in configuration.")
         run_args.update(runs[run_name])
     else:
-        run_name = 'default'
+        run_name = "default"
 
     # Process databases + tables
-    for index in '12':
+    for index in "12":
         args = run_args.pop(index, {})
-        for attr in ('database', 'table'):
+        for attr in ("database", "table"):
             if attr not in args:
                 raise ConfigParseError(f"Running 'run.{run_name}': Connection #{index} in missing attribute '{attr}'.")
 
-        database = args.pop('database')
-        table = args.pop('table')
-        threads = args.pop('threads', None)
+        database = args.pop("database")
+        table = args.pop("table")
+        threads = args.pop("threads", None)
         if args:
             raise ConfigParseError(f"Unexpected attributes for connection #{index}: {args}")
 
         if not is_uri(database):
             if database not in databases:
-                raise ConfigParseError(f"Database '{database}' not found in list of databases. Available: {list(databases)}.")
+                raise ConfigParseError(
+                    f"Database '{database}' not found in list of databases. Available: {list(databases)}."
+                )
             database = dict(databases[database])
             assert isinstance(database, dict)
-            if 'driver' not in database:
+            if "driver" not in database:
                 raise ConfigParseError(f"Database '{database}' did not specify a driver.")
 
-        run_args[f'database{index}'] = database
-        run_args[f'table{index}'] = table
+        run_args[f"database{index}"] = database
+        run_args[f"table{index}"] = table
         if threads is not None:
-            run_args[f'threads{index}'] = int(threads)
+            run_args[f"threads{index}"] = int(threads)
 
     # Update keywords
-    new_kw = dict(kw)           # Set defaults
-    new_kw.update(run_args)     # Apply config
-    new_kw.update({k:v for k, v in kw.items() if v})   # Apply non-empty defaults
+    new_kw = dict(kw)  # Set defaults
+    new_kw.update(run_args)  # Apply config
+    new_kw.update({k: v for k, v in kw.items() if v})  # Apply non-empty defaults
 
-    new_kw['__conf__'] = run_args
+    new_kw["__conf__"] = run_args
 
     return new_kw
+
+
+def apply_config_from_file(path: str, run_name: str, kw: Dict[str, Any]):
+    with open(path) as f:
+        return _apply_config(toml.load(f), run_name, kw)
+
+
+def apply_config_from_string(toml_config: str, run_name: str, kw: Dict[str, Any]):
+    return _apply_config(toml.loads(toml_config), run_name, kw)

data_diff/databases/connect.py

@@ -137,9 +137,10 @@ def connect_to_uri(db_uri: str, thread_count: Optional[int] = 1) -> Database:
 
     return cls(**kw)
 
+
 def connect_with_dict(d, thread_count):
     d = dict(d)
-    driver = d.pop('driver')
+    driver = d.pop("driver")
     try:
         matcher = MATCH_URI_PATH[driver]
     except KeyError:
@@ -151,6 +152,7 @@ def connect_with_dict(d, thread_count):
 
     return cls(**d)
 
+
 def connect(x, thread_count):
     if isinstance(x, str):
         return connect_to_uri(x, thread_count)

data_diff/databases/oracle.py

@@ -27,7 +27,7 @@ class Oracle(ThreadedDatabase):
     def __init__(self, *, host, database, thread_count, **kw):
         self.kwargs = dict(dsn="%s/%s" % (host, database), **kw)
 
-        self.default_schema = kw.get('user')
+        self.default_schema = kw.get("user")
 
         super().__init__(thread_count=thread_count)
 

data_diff/diff_tables.py

@@ -271,7 +271,7 @@ class TableDiffer:
     """
 
     bisection_factor: int = DEFAULT_BISECTION_FACTOR
-    bisection_threshold: Number = DEFAULT_BISECTION_THRESHOLD       # Accepts inf for tests
+    bisection_threshold: Number = DEFAULT_BISECTION_THRESHOLD  # Accepts inf for tests
     threaded: bool = True
     max_threadpool_size: Optional[int] = 1
 

tests/test_config.py

@@ -0,0 +1,43 @@
+import unittest
+
+from data_diff.config import apply_config_from_string, ConfigParseError
+
+
+class TestConfig(unittest.TestCase):
+    def test_basic(self):
+        config = r"""
+            [database.test_postgresql]
+            driver = "postgresql"
+            user = "postgres"
+            password = "Password1"
+
+            [run.default]
+            update_column = "timestamp"
+            verbose = true
+            threads = 2
+
+            [run.pg_pg]
+            threads = 4
+            1.database = "test_postgresql"
+            1.table = "rating"
+            1.threads = 11
+            2.database = "postgresql://postgres:Password1@/"
+            2.table = "rating_del1"
+            2.threads = 22
+        """
+        self.assertRaises(ConfigParseError, apply_config_from_string, config, "bla", {})  # No such run
+
+        res = apply_config_from_string(config, "pg_pg", {})
+        assert res["update_column"] == "timestamp"  # default
+        assert res["verbose"] is True
+        assert res["threads"] == 4  # overwritten by pg_pg
+        assert res["database1"] == {"driver": "postgresql", "user": "postgres", "password": "Password1"}
+        assert res["database2"] == "postgresql://postgres:Password1@/"
+        assert res["table1"] == "rating"
+        assert res["table2"] == "rating_del1"
+        assert res["threads1"] == 11
+        assert res["threads2"] == 22
+
+        res = apply_config_from_string(config, "pg_pg", {"update_column": "foo", "table2": "bar"})
+        assert res["update_column"] == "foo"
+        assert res["table2"] == "bar"