initial commit

conftest.py

@@ -0,0 +1,55 @@
+import subprocess as sp
+import paramiko
+import pytest
+
+from os import getenv
+
+@pytest.fixture
+def server():
+    try:
+        s = paramiko.SSHClient()
+        s.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+
+        s.connect(hostname = getenv("host"),
+                  username = getenv("user"),
+                  password = getenv("pass"))
+
+        print("con suc")
+
+        _, _, _ = s.exec_command("iperf3 -s")
+
+        yield True
+
+    except paramiko.AuthenticationException:
+        print("aut fal")
+        yield False
+    except paramiko.SSHException as e:
+        print(f"err {e}")
+        yield False
+    finally:
+        s.close()
+
+@pytest.fixture
+def client(server):
+    assert server
+
+    res = ""
+
+    p = sp.Popen(
+            ["iperf3", "--forceflush", "-i", "1", "-c", getenv("host")],
+            stdout = sp.PIPE,
+            stderr = sp.PIPE)
+
+    while True:
+        l = p.stdout.readline()
+
+        if l:
+            res += l.decode("UTF-8")
+
+        if p.poll() != None:
+            break
+
+    if p.stderr.read():
+        return False
+    else:
+        return res

iperf_test.py

@@ -0,0 +1,12 @@
+from parser import get_iperf3_stats
+
+class TestSuite():
+    def test_conn_speed(self, client):
+        assert client != False
+
+        s = get_iperf3_stats(client)
+
+        assert len(s) == 10
+
+        for i in s:
+            assert i['bitrate'] >= 80_000_000 # 80+ Mbit/s

parser.py

@@ -0,0 +1,31 @@
+def get_iperf3_stats(s):
+    pool = s.split("\n")
+
+    stats = []
+
+    interpret_stats = False
+
+    for i in pool:
+        c = i.split()
+
+        if 'ID]' in c:
+            interpret_stats = True
+            continue
+
+        if '-' in c:
+            interpret_stats = False
+            break
+
+        if interpret_stats:
+            if len(c) < 8:
+                continue
+
+            stat = {}
+
+            stat['int_start'], stat['int_end'] = list(map(float, c[2].split('-')))
+            stat['transfer'] = float(c[4]) * 1024 ** ['B', 'K', 'M', 'G'].index(c[5][0])
+            stat['bitrate'] = float(c[6]) * 1024 ** ['b', 'K', 'M', 'G'].index(c[7][0])
+
+            stats.append(stat)
+
+    return stats