diff --git a/.gitignore b/.gitignore index d3efeb9..36345c7 100644 --- a/.gitignore +++ b/.gitignore @@ -150,7 +150,6 @@ dmypy.json *.bak *.gho *.ori -*.orig *.tmp ### Windows template @@ -202,7 +201,6 @@ export_presets.cfg *.bz2 *.xz *.lzma -*.cab # Packing-only formats *.iso @@ -212,15 +210,10 @@ export_presets.cfg *.dmg *.xpi *.gem -*.egg *.deb *.rpm -*.msi -*.msm -*.msp ### Patch template -*.orig *.rej ### macOS template @@ -256,7 +249,6 @@ Temporary Items *.exe *.exe~ *.dll -*.so *.dylib # Test binary, built with `go test -c` @@ -271,7 +263,6 @@ Temporary Items ### VirtualEnv template # Virtualenv # http://iamzed.com/2009/05/07/a-primer-on-virtualenv/ -.Python [Bb]in [Ii]nclude [Ll]ib @@ -279,7 +270,6 @@ Temporary Items [Ll]ocal [Ss]cripts pyvenv.cfg -.venv pip-selfcheck.json ### PuTTY template @@ -305,6 +295,6 @@ tags # Persistent undo [._]*.un~ -/.env /test-results -/.idea \ No newline at end of file +/.idea/workspace.xml +/.idea/tasks.xml diff --git a/tests/test_poc.py b/tests/test_poc.py index 986c2a3..5aa0efe 100644 --- a/tests/test_poc.py +++ b/tests/test_poc.py @@ -1,41 +1,38 @@ -from jinja2 import Environment -import urllib.request - import random - import socket - -import sys -from time import sleep -import splunklib.results as results -import splunklib.client as client +import urllib.request from time import sleep + import pytest +import splunklib.client as client +from jinja2 import Environment env = Environment(extensions=['jinja2_time.TimeExtension']) + @pytest.fixture def setup_wordlist(): - word_url = "http://svnweb.freebsd.org/csrg/share/dict/words?view=co&content-type=text/plain" response = urllib.request.urlopen(word_url) long_txt = response.read().decode() return long_txt.splitlines() + @pytest.fixture def setup_splunk(): - tried=0 + tried = 0 while True: try: c = client.connect(username="admin", password="Changed@11", host="splunk", port="8089") break except ConnectionRefusedError: - tried +=1 - if tried>90: + tried += 1 + if tried > 90: raise sleep(1) return c + def sendsingle(message): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_address = ('sc4s', 514) @@ -46,14 +43,15 @@ def sendsingle(message): sock.connect(server_address) break except: - tried +=1 + tried += 1 if tried > 90: raise sleep(1) sock.sendall(str.encode(message)) sock.close() -def splunk_single(service,search): + +def splunk_single(service, search): kwargs_normalsearch = {"exec_mode": "normal"} tried = 0 while True: @@ -76,7 +74,7 @@ def splunk_single(service,search): # Get the results and display them resultCount = stats["resultCount"] eventCount = stats["eventCount"] - if resultCount > 0 or tried>15: + if resultCount > 0 or tried > 15: break else: tried += 1 @@ -84,7 +82,7 @@ def splunk_single(service,search): return resultCount, eventCount -def test_defaultroute(record_property,setup_wordlist,setup_splunk): +def test_defaultroute(record_property, setup_wordlist, setup_splunk): host = "{}-{}".format(random.choice(setup_wordlist), random.choice(setup_wordlist)) mt = env.from_string("{{ mark }} {% now 'utc', '%b %d %H:%M:%S' %}.000z {{ host }} sc4s_default[0]: test\n") @@ -92,15 +90,13 @@ def test_defaultroute(record_property,setup_wordlist,setup_splunk): sendsingle(message) - st= env.from_string("search \"{{ host }}\" | head 2") + st = env.from_string("search \"{{ host }}\" | head 2") search = st.render(host=host) - resultCount, eventCount = splunk_single(setup_splunk,search) + resultCount, eventCount = splunk_single(setup_splunk, search) record_property("host", host) record_property("resultCount", resultCount) record_property("message", message) assert resultCount == 1 - -