Test/integrations (#91)

* Added initial setup for integration tests

* Integration-test expects a profile set with password

* Change tox to execute only unit tests

* Refactor- use context management for teardown

* Refactor- added teardown decorator

* Add more alert tests

* Added documentation

* Use pexpect for prompt

* Fix response change in pexpect

* Added assert statement to validate non-empty response

* Receiving EOF should be considered success

* Fix single line response

* make cleanup task OS independent for integration test

Author: kiran-chaudhary

integration/__init__.py

@@ -0,0 +1,33 @@
+import os
+import pexpect
+
+
+LINE_FEED = b'\r\n'
+PASSWORD_PROMPT = b'Password: '
+ENCODING_TYPE = 'utf-8'
+
+
+def encode_response(line, encoding_type=ENCODING_TYPE):
+    return line.decode(encoding_type)
+
+
+def run_command(command):
+
+    process = pexpect.spawn(command)
+    response = []
+    try:
+        expected = process.expect([PASSWORD_PROMPT, pexpect.EOF])
+        if expected == 0:
+            process.sendline(os.environ["C42_PW"])
+            process.expect(LINE_FEED)
+            output = process.readlines()
+            response = [encode_response(line) for line in output]
+        else:
+            output = process.before
+            response = encode_response(output).splitlines()
+    except pexpect.TIMEOUT:
+        return 1, response
+    return 0, response
+
+
+__all__ = [run_command]

integration/test_alerts.py

@@ -0,0 +1,67 @@
+import pytest
+import json
+
+from integration import run_command
+from integration.util import cleanup_after_validation
+
+ALERT_COMMAND = "code42 alerts print -b 2020-05-18 -e 2020-05-20"
+
+
+def _parse_response(response):
+    return [json.loads(line) for line in response if len(line)]
+
+
+def _validate_field_value(field, value, response):
+    parsed_response = _parse_response(response)
+    assert len(parsed_response) > 0
+    for record in parsed_response:
+        assert record[field] == value
+
+
+@pytest.mark.parametrize(
+    "command, field, value",
+    [("{} --state OPEN".format(ALERT_COMMAND), "state", "OPEN"),
+     ("{} --state RESOLVED".format(ALERT_COMMAND), "state", "RESOLVED"),
+     ("{} --actor spatel@code42.com".format(ALERT_COMMAND), "actor", "spatel@code42.com"),
+     ("{} --rule-name 'File Upload Alert'".format(ALERT_COMMAND), "name", "File Upload Alert"),
+     ("{} --rule-id 962a6a1c-54f6-4477-90bd-a08cc74cbf71".format(ALERT_COMMAND), "ruleId",
+      "962a6a1c-54f6-4477-90bd-a08cc74cbf71"),
+     ("{} --rule-type FedEndpointExfiltration".format(ALERT_COMMAND), "type",
+      "FED_ENDPOINT_EXFILTRATION"),
+     ("{} --description 'Alert on any file upload'".format(ALERT_COMMAND), "description",
+      "Alert on any file upload events"),
+     ]
+)
+def test_alert_prints_to_stdout_and_filters_result_by_given_value(command, field, value):
+    return_code, response = run_command(command)
+    assert return_code is 0
+    _validate_field_value(field, value, response)
+
+
+def _validate_begin_date(response):
+    parsed_response = _parse_response(response)
+    assert len(parsed_response) > 0
+    for record in parsed_response:
+        assert record["createdAt"].startswith("2020-05-18")
+
+
+@pytest.mark.parametrize("command, validate", [
+    (ALERT_COMMAND, _validate_begin_date),
+])
+def test_alert_prints_to_stdout_and_filters_result_between_given_date(command, validate):
+    return_code, response = run_command(command)
+    assert return_code is 0
+    validate(response)
+
+
+def _validate_severity(response):
+    record = json.loads(response)
+    assert record["severity"] == "MEDIUM"
+
+
+@cleanup_after_validation("./integration/alerts")
+def test_alert_writes_to_file_and_filters_result_by_severity():
+    command = "code42 alerts write-to ./integration/alerts -b 2020-05-18 -e 2020-05-20 " \
+              "--severity MEDIUM"
+    return_code, response = run_command(command)
+    return _validate_severity

integration/util.py

@@ -0,0 +1,29 @@
+import os
+
+
+class cleanup(object):
+    def __init__(self, filename):    
+        self.filename = filename
+
+    def __enter__(self):
+        return open(self.filename, "r")
+    
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        os.remove(self.filename)
+
+
+def cleanup_after_validation(filename):
+    """Decorator to read response from file for `write-to` commands and cleanup the file after test
+    execution.
+
+    The decorated function should return validation function that takes the content of the file
+    as input. e.g `test_alerts.py::test_alert_writes_to_file_and_filters_result_by_severity`
+    """
+    def wrap(test_function):
+        def wrapper():
+            validate = test_function()
+            with cleanup(filename) as f:
+                response = f.read()
+                validate(response)
+        return wrapper
+    return wrap

run_integration.py

@@ -0,0 +1,11 @@
+import sys
+import os
+
+if __name__ == "__main__":
+    if sys.argv[1] and sys.argv[2]:
+        os.environ["C42_USER"] = sys.argv[1]
+        os.environ["C42_PW"] = sys.argv[2]
+        rc = os.system("pytest ./integration -v -rsxX -l --tb=short --strict")
+        sys.exit(rc)
+    else:
+        print("username and password were not supplied. Integration tests will be skipped.")

setup.py

@@ -25,6 +25,7 @@
         "keyring==18.0.1",
         "keyrings.alt==3.2.0",
         "py42>=1.2.0",
+        "pexpect>=4.8"
     ],
     license="MIT",
     include_package_data=True,

tox.ini

@@ -18,7 +18,7 @@ commands =
     # -l: show locals in tracebacks
     # --tb=short: short traceback print mode
     # --strict: marks not registered in configuration file raise errors
-    pytest --cov=code42cli --cov-append  -v -rsxX -l --tb=short --strict --disable-pytest-warnings
+    pytest tests --cov=code42cli --cov-append  -v -rsxX -l --tb=short --strict --disable-pytest-warnings
 
 depends =
     {py35,py36,py37,py38}: clean