Merge pull request #4 from getsentry/iw/apply-on-update

Adding implementation for apply_on_update

lib/sh.py

@@ -1,12 +1,21 @@
 from __future__ import annotations
 
 import contextlib
+import os
 from os import getenv
 from typing import TYPE_CHECKING
+from typing import Dict
 from typing import Iterable
 from typing import Iterator
+from typing import List
+from typing import MutableMapping
+from typing import Union
 
 US_ASCII = "US-ASCII"  # the least-ambiguous encoding
+JSONPrimitive = Union[str, int, float, bool, None]
+JSONObject = Dict[str, JSONPrimitive]
+JSONArray = List[JSONPrimitive]
+JSONValue = Union[JSONPrimitive, JSONArray, JSONObject]
 
 if TYPE_CHECKING:
     # strict encapsulation: limit run-time access to just one function each
@@ -37,11 +46,25 @@ def xtrace(cmd: Command) -> None:
         info((PS4, quote(cmd)))
 
 
-def cd(dirname: str) -> None:
+def cd(
+    dirname: str,
+    env: MutableMapping[str, str] = os.environ,
+    direnv: bool = True,
+) -> None:
     from os import chdir
 
     xtrace(("cd", dirname))
     chdir(dirname)
+    if direnv:
+        direnv_json: JSONValue = json(("direnv", "export", "json"))
+        if not isinstance(direnv_json, dict):
+            raise TypeError(f"expected dict, got {type(direnv_json)}")
+        for key, value in direnv_json.items():
+            if value is None:
+                env.pop(key, None)
+            else:
+                assert isinstance(value, str), value
+                env[key] = value
 
 
 def quote(cmd: Command) -> str:
@@ -54,8 +77,16 @@ def stdout(cmd: Command) -> str:
     return _wait(_popen(cmd, capture_output=True)).stdout.rstrip("\n")
 
 
-def json(cmd: Command, encoding: str = US_ASCII) -> Iterable[object]:
-    """Return a parsing of newline-delimited json on a subprocess' stdout."""
+def json(cmd: Command) -> JSONValue:
+    """Parse the (singular) json on a subprocess' stdout."""
+    import json
+
+    result: JSONValue = json.loads(stdout(cmd))
+    return result
+
+
+def jq(cmd: Command, encoding: str = US_ASCII) -> Iterable[object]:
+    """Yield the objects from newline-delimited json on a subprocess' stdout."""
     import json
 
     process = _popen(cmd, encoding=encoding, capture_output=True)

manual_tests/behaviors/apply_on_update.py

@@ -1,7 +1,33 @@
 #!/usr/bin/env py.test
 from __future__ import annotations
 
+from lib.functions import now
+from manual_tests.lib import gh
+from manual_tests.lib import gha
+from manual_tests.lib import slice
+from manual_tests.lib import tacos_demo
+from manual_tests.lib import tf
+
+TEST_NAME = __name__
+
+Branch = int
+
 
 def test() -> None:
-    # TODO: anything
-    pass
+    tacos_demo.clone()
+    tacos_demo_pr = tacos_demo.new_pr(TEST_NAME, slice.random())
+    try:
+        since = now()
+        gha.assert_eventual_success("terraform_lock", since)
+
+        gh.approve_pr(tacos_demo_pr.url)
+        gh.assert_pr_is_approved(tacos_demo_pr.url)
+
+        # the taco-apply label causes the plan to become clean:
+        assert not tf.plan_clean()
+        since = now()
+        gh.add_label(tacos_demo_pr.url, ":taco::apply")
+        gha.assert_eventual_success("terraform_apply", since)
+        assert tf.plan_clean()
+    finally:
+        gh.close_pr(tacos_demo_pr.branch)

manual_tests/lib/gh.py

@@ -14,7 +14,7 @@
 
 def assert_matching_comment(comment: str, since: datetime) -> None:
     # Fetch the comments on the PR
-    comments = sh.json(
+    comments = sh.jq(
         (
             "gh",
             "pr",
@@ -41,6 +41,30 @@ def assert_matching_comment(comment: str, since: datetime) -> None:
         raise AssertionError(f"No matching comment: {comment}\n{comments}")
 
 
+def assert_pr_has_label(pr_url: URL, label: str) -> None:
+    sh.banner("asserting PR has label:")
+    labels = sh.stdout(
+        (
+            "gh",
+            "pr",
+            "view",
+            "--json",
+            "labels",
+            "--jq",
+            ".labels.[] | .name",
+            pr_url,
+        )
+    )
+
+    assert label in labels, (label, labels)
+
+
+def assert_pr_is_approved(pr_url: URL) -> None:
+    sh.banner("asserting PR is approved:")
+    # TODO: actually check that the PR is approved (once we add a second service account)
+    assert_pr_has_label(pr_url, ":taco::approve")
+
+
 def open_pr(branch: Branch) -> str:
     return sh.stdout(("gh", "pr", "create", "--fill-first", "--head", branch))
 
@@ -66,3 +90,14 @@ def close_pr(pr_url: URL) -> None:
             pr_url,
         )
     )
+
+
+def approve_pr(pr_url: URL) -> None:
+    sh.banner("approving PR:")
+    # TODO: find a way to approve with a separate service account
+    add_label(pr_url, ":taco::approve")
+
+
+def add_label(pr_url: URL, label: str) -> None:
+    sh.banner(f"adding label {label} to PR:")
+    sh.run(("gh", "pr", "edit", "--add-label", label, pr_url))

manual_tests/lib/gha.py

@@ -12,7 +12,7 @@
 
 def get_checks() -> Iterable[object]:
     """get the most recent run of the named check"""
-    return sh.json(
+    return sh.jq(
         (
             "gh",
             "pr",

manual_tests/lib/tf.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from lib import sh
+
+
+def plan_clean() -> bool:
+    """
+    Returns whether running terraform plan differs from the current state
+    """
+    return sh.success(
+        ("sudo-sac", "terragrunt", "run-all", "plan", "--detailed-exitcode")
+    )