1import sys
 2import os
 3import sh
 4from sh import psql
 5import pytest
 6import subprocess
 7import psycopg2.extras
 8from pathlib import Path
 9from pprint import pprint
10
11
12test_dir = Path(Path(__file__).parent)
13
14
15@pytest.fixture
16def chartepssg_bin(project_root):
17    return sh.Command(sys.executable).bake(project_root / "bin/chartepssg")
18
19
20@pytest.fixture
21def postgresql_db_chart(psql_conn):
22<<<hidden due to potential security issue>>>
23    psql(
24        "--host=localhost",
25        "--port=5123",
26        "--username=dev",
27        "-d",
28        "postgres",
29        "-f",
30        test_dir / "testdata/chartepssg_recreate.sql",
31    )
32    return psql_conn
33
34
35@pytest.mark.parametrize(
36    "inputfile,extractfunc,expectednames,table",
37    [
38        (
39            test_dir / "testdata/5_tc_packets.rapid",
40            lambda x: x["name"],
41            ["CYC12008", "ASC17000", "ASC17000", "ASC17000", "ASC17000"],
42            "tc_store",
43        ),
44        (
45            test_dir / "testdata/5_tm_packets.rapid",
46            lambda x: list(x["payload"].keys())[0][:3],
47            ["MIS", "CFT", "CFT", "CST"],
48            "tm",
49        ),
50    ],
51)
52def test_ccsds_ingestion_tm_tc(
53    inputfile, extractfunc, expectednames, table, postgresql_db_chart, chartepssg_bin
54):
55    chartepssg_bin(
56        "ccsds", "--db", "integrationtest", "-s", "EPSSGA", inputfile, "--ingest"
57    )
58
59    cursor = postgresql_db_chart.cursor(cursor_factory=psycopg2.extras.DictCursor)
60
61    cursor.execute(f"select * from {table}")
62    ret = [dict(x) for x in cursor.fetchall()]
63    pprint(ret)
64    assert len(ret) == len(expectednames)
65
66    names = [extractfunc(x) for x in ret]
67    assert names == expectednames
68
69
70def test_ccsds_ingestion_onboard_events(postgresql_db_chart, chartepssg_bin):
71    inputfile = test_dir / "testdata/5_tm_onboardevent_packets.rapid"
72    chartepssg_bin(
73        "ccsds", "--db", "integrationtest", "-s", "EPSSGA", inputfile, "--ingest"
74    )
75
76    print(postgresql_db_chart.get_dsn_parameters())
77    cursor = postgresql_db_chart.cursor(cursor_factory=psycopg2.extras.DictCursor)
78
79    cursor.execute("select * from tm")
80    ret = [dict(x) for x in cursor.fetchall()]
81    assert len(ret) == 5  # one packet is not in the MIB
82
83    cursor.execute("select * from events")
84    ret = [dict(x) for x in cursor.fetchall()]
85    assert len(ret) == 0
86
87    chartepssg_bin(
88        "alg",
89        "ONBOARD_EVENTS",
90        "--sensing-start=2019-01-01T00:00:00Z",
91        "--sensing-stop=2021-05-01T00:00:00Z",
92        "--db=integrationtest",
93        "--sid=epssga",
94        "--ingest",
95    )
96
97    cursor.execute("select * from events")
98    ret = [dict(x) for x in cursor.fetchall()]
99    assert len(ret) == 5