ksconf promote: Fixed empty stanza bug

- Resolved bug where empty stanzas in the local file could result in deletion
  of a populated stanza in default.
- Revamped diff handling internally.

.gitignore

@@ -2,6 +2,7 @@
 *.swp
 
 .idea
+.vscode
 .vagrant
 
 # Autogenerated by the build system

docs/source/changelog.rst

@@ -8,6 +8,11 @@ Ksconf 0.7.x
 
 New functionality, massive documentation improvements, metadata support, and Splunk app install fixes.
 
+Release v0.7.10 (DRAFT)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+-  Fixed bug where empty stanzas in the local file could result in deletion in default with ``ksconf promote``.
+   Updated diff interface to improve handling of empty stanzas, but wider support is still needed across other commands; but this isn't a high priority.
 
 Release v0.7.9 (2020-09-23)
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~

ksconf/commands/promote.py

@@ -63,6 +63,11 @@
 
 
 
+def empty_dict(d):
+    # Or just   d == {}?   Not sure which is better
+    return isinstance(d, dict) and len(d) == 0
+
+
 class PromoteCmd(KsconfCmd):
     help = dedent("""\
     Promote .conf settings between layers using either batch or interactive mode.
@@ -305,6 +310,14 @@ def prep_filters(self, args):
         if args.stanza:
             return True
 
+    @staticmethod
+    def combine_stanza(a, b):
+        if a is None:
+            return b
+        d = dict(a)
+        d.update(b)
+        return d
+
     def _do_promote_automatic(self, cfg_src, cfg_tgt, args):
         # Promote ALL entries;  simply, isn't it...  ;-)
         final_cfg = merge_conf_dicts(cfg_tgt, cfg_src)
@@ -400,9 +413,16 @@ def prompt_yes_no(prompt):
                 '''
                 if isinstance(op.location, DiffStanza):
                     # Move entire stanza
-                    show_diff(self.stdout, [op])
+                    """ ----  If we need to support empty stanza promotion.....
+                    print("OP:   {!r}".format(op))
+                    if empty_dict(op.b):
+                        print("Empty stanza [{}]".format(op.location.stanza))
+                    else:
+                    """
+                    if True:
+                        show_diff(self.stdout, [op])
                     if prompt_yes_no("Apply [{0}]".format(op.location.stanza)):
-                        out_cfg[op.location.stanza] = op.b
+                        out_cfg[op.location.stanza] = self.combine_stanza(op.a, op.b)
                         del out_src[op.location.stanza]
                 else:
                     show_diff(self.stdout, [op])
@@ -427,7 +447,7 @@ def _do_promote_list(self, cfg_src, cfg_tgt, args):
                     show_diff(self.stdout, [op])
                 if isinstance(op.location, DiffStanza):
                     # Move entire stanza
-                    out_cfg[op.location.stanza] = op.b
+                    out_cfg[op.location.stanza] = self.combine_stanza(op.a, op.b)
                     del out_src[op.location.stanza]
                 else:
                     # Move key

ksconf/conf/delta.py

@@ -49,13 +49,18 @@ def __str__(self):
         return "{0:50} {1}".format(self.name, ts)
 
 
-def compare_stanzas(a, b, stanza_name):
+def compare_stanzas(a, b, stanza_name, preserve_empty=False):
+    if preserve_empty:
+        is_empty = lambda v: v is None
+    else:
+        is_empty = lambda v: not v
+
     if a == b:
         return [DiffOp(DIFF_OP_EQUAL, DiffStanza("stanza", stanza_name), a, b) ]
-    elif not b:
+    elif is_empty(b):
         # A only
         return [ DiffOp(DIFF_OP_DELETE, DiffStanza("stanza", stanza_name), a, None) ]
-    elif not a:
+    elif is_empty(a):
         # B only
         return [ DiffOp(DIFF_OP_INSERT, DiffStanza("stanza", stanza_name), None, b) ]
     else:
@@ -84,7 +89,7 @@ def _compare_stanzas(a, b, stanza_name):
             yield DiffOp(DIFF_OP_REPLACE, DiffStzKey("key", stanza_name, key), a_, b_)
 
 
-def compare_cfgs(a, b, allow_level0=True):
+def compare_cfgs(a, b, allow_level0=True, preserve_empty=False):
     '''
     Return list of 5-tuples describing how to turn a into b.
 
@@ -147,18 +152,7 @@ def compare_cfgs(a, b, allow_level0=True):
         all_stanzas = list(all_stanzas)
     all_stanzas = sorted(all_stanzas)
     for stanza in all_stanzas:
-        if stanza in a and stanza in b:
-            if a[stanza] == b[stanza]:
-                delta.append(DiffOp(DIFF_OP_EQUAL, DiffStanza("stanza", stanza),
-                                    a[stanza], b[stanza]))
-            else:
-                delta.extend(_compare_stanzas(a[stanza], b[stanza], stanza))
-        elif stanza in a:
-            # A only
-            delta.append(DiffOp(DIFF_OP_DELETE, DiffStanza("stanza", stanza), a[stanza], None))
-        else:
-            # B only
-            delta.append(DiffOp(DIFF_OP_INSERT, DiffStanza("stanza", stanza), None, b[stanza]))
+        delta.extend(compare_stanzas(a.get(stanza), b.get(stanza), stanza, preserve_empty))
     return delta
 
 

tests/test_cli_promote.py

@@ -188,6 +188,29 @@ def test_promote_list_add_new_stanza_ignore_case(self):
             self.assertNotIn("Stanza2", l)
         del twd
 
+    def test_promote_empty_stanza(self):
+        twd = TestWorkDir()
+        conf_default = twd.write_file("default/test.conf", r"""
+        [Stanza1]
+        a = 2
+        c = 8
+        [Stanza3]
+        z = 0
+        """)
+        conf_local = twd.write_file("local/test.conf", r"""
+        [Stanza1]
+
+        [Stanza3]
+        z = 2
+        """)
+        with ksconf_cli:
+            ksconf_cli("promote", "--batch", conf_local, conf_default)
+            d = twd.read_conf("default/test.conf")
+            self.assertFalse(os.path.isfile(conf_local))
+            self.assertEqual(d["Stanza1"]["a"], "2")
+            self.assertEqual(d["Stanza1"]["c"], "8")
+            self.assertEqual(d["Stanza3"]["z"], "2")
+
     def test_promote_batch_simple_keep(self):
         twd = self.sample_data01()
         with ksconf_cli:

tests/test_parser.py

@@ -491,7 +491,7 @@ def test_empty_stanzas(self):
 
         [in_b_only]
         """)
-        delta = compare_cfgs(a, b)
+        delta = compare_cfgs(a, b, preserve_empty=True)
         delta_search = partial(self.find_op_by_location, delta)
         self.assertEqual(delta_search("stanza", stanza="common").tag, DIFF_OP_EQUAL)
         self.assertEqual(delta_search("stanza", stanza="in_a_only").tag, DIFF_OP_DELETE)