Python - Functie wordt niet geimporteerd..

[Heaget]

Beste Tweakers,

Ik heb denk ik te lang naar mijn eigen code zitten staren, maar ik zie echt het probleem niet. Waarschijnlijk is het iets doms en heb ik na de eerste comment een "ja, natuurlijk dat is het"-moment, maar nu zie ik het even niet.

Ik heb 3 python bestanden. 1 main bestand, 1 test bestand voor pytest, en 1 bestand waar een simpele functie in staat.

Het testbestand en het functiebestand werken goed samen. Bovenin het testbestand heb ik from functie import countTcpSyn staan, de tests komen eruit zoals verwacht.

Nu wil ik in het main bestand deze functie op dezelfde wijze importeren, maar wat ik ook doe, het wil niet werken.

Hoe zorg ik ervoor dat mijn functie uit functie.py wordt geimporteerd in mijn main bestand, en deze uitvoert op de plek waar nu de code zelf nog staat, maar dan uiteraard met het gekozen JSON bestand?

Main bestand

code:

import json
import heapq
import argparse

# Create an argument parser
parser = argparse.ArgumentParser(description='Update dataset.json with a new file.')

# Add the file argument
parser.add_argument('file_path', type=str, nargs='?', default='dataset.json', help='Path to the file to update dataset.json with.')
parser.add_argument('-p', '--protocol', action='store_true', help='display protocols used in JSON file' )
parser.add_argument('-pc', '--packetcount', action='store_true', help='display packetcount in JSON file' )
parser.add_argument('-tcp', '--tcpflags', action='store_true', help='''display top 3 IP's with SYN bits on 1 ''' )

# Parse the command-line arguments
args = parser.parse_args()

# Load the JSON data from the provided file
with open(args.file_path) as file:
    data = json.load(file)

class JSONAnalyzer:
    def __init__(self, file_path='dataset.json'):
        self.file_path = file_path
        self.data = None

    def load_data(self):
        with open(self.file_path) as file:
            self.data = json.load(file)

    def display_protocols(self):
        protocols = self.data[0]['_source']['layers']['frame']['frame.protocols']
        protocol_list = protocols.split(':')
        print("These protocols are being used:", protocol_list)

    def display_packet_counts(self):
        packet_counts = {}

        for packet in self.data:
            timestamp = packet["_source"]["layers"]["frame"]["frame.time"]
            minute = timestamp.split(':')[1]
            packet_counts[minute] = packet_counts.get(minute, 0) + 1

        for minute, count in packet_counts.items():
            print(f"Minute {minute}: {count} packets")

    def display_top_tcp_flags(self):
        def count_tcp_syn(data):
            ip_src_syn_count = {}

            for packet in data:
                layers = packet["_source"]["layers"]
                ip_src = layers["ip"]["ip.src"]
                tcp_flags_syn = layers["tcp"]["tcp.flags_tree"]["tcp.flags.syn"]

                if ip_src in ip_src_syn_count:
                    ip_src_syn_count[ip_src] += int(tcp_flags_syn)
                else:
                    ip_src_syn_count[ip_src] = int(tcp_flags_syn)

            return ip_src_syn_count

        ip_src_syn_count = count_tcp_syn(self.data)
        top_counts = heapq.nlargest(3, ip_src_syn_count.values())

        print("Top 3 IP source and tcp.flags.syn counts (descending order):")
        for syn_count in top_counts:
            for ip_src, count in ip_src_syn_count.items():
                if count == syn_count:
                    print("IP source:", ip_src)
                    print("Number of TCP connection requests by tcp.flags.syn = 1:", count)
                    print()

# Create an instance of JSONAnalyzer
analyzer = JSONAnalyzer()

# Load the JSON data from the file
analyzer.load_data()

# Use the methods to analyze and display information
if args.protocol == True:
    analyzer.display_protocols()
if args.packetcount == True:
    analyzer.display_packet_counts()
if args.tcpflags == True:
    analyzer.display_top_tcp_flags()

Test bestand

code:

from functie import countTcpSyn

def test_countTcpSyn():
    # Define a sample data input for the function
    data = [
        {
            "_source": {
                "layers": {
                    "ip": {"ip.src": "192.168.0.1"},
                    "tcp": {"tcp.flags_tree": {"tcp.flags.syn": "1"}}
                }
            }
        },
        {
            "_source": {
                "layers": {
                    "ip": {"ip.src": "192.168.1.9"},
                    "tcp": {"tcp.flags_tree": {"tcp.flags.syn": "1"}}
                }
            }
        },
        {
            "_source": {
                "layers": {
                    "ip": {"ip.src": "192.168.0.1"},
                    "tcp": {"tcp.flags_tree": {"tcp.flags.syn": "1"}}
                }
            }
        }
    ]

    # Call the function to get the result
    result = countTcpSyn(data)

    # Assert the expected output based on the sample data
    expected_result = {"192.168.0.1": 2, "192.168.1.9": 1,}
    assert result == expected_result

Functie

code:

def countTcpSyn(data):
    ipSrcSynCount = {}
    # For loop that counts the tcp.flags.syn bit per IP source adres
    for packet in data:
        layers = packet["_source"]["layers"]
        ipSrc = layers["ip"]["ip.src"]
        tcpFlagsSyn = layers["tcp"]["tcp.flags_tree"]["tcp.flags.syn"]

        if ipSrc in ipSrcSynCount:
            ipSrcSynCount[ipSrc] += int(tcpFlagsSyn)
        else:
            ipSrcSynCount[ipSrc] = int(tcpFlagsSyn)

    return ipSrcSynCount

[cosmo_roel]

code:

from .functie

of

code:

from jouwpackagenaam.functie

[ouweklimgeit]

Bovenaan je main gewoon;

code:

import test, functie

Als je vervolgens de def's() nodig hebt, dan gebruik je in de Main gewoon

code:

test.test_countTcpSyn()

of

code:

functie.countTcpSyn()

Verder zou ik in het testbestand gewoon een return gebruiken om de json terug te gooien richting de Main, dan hoef je in die functie ook geen extra imports te gebruiken.

code:

main.py

jsondata=test._countTcpSyn()
result=functie.countTcpSyn(jsondata)

[ Voor 3% gewijzigd door ouweklimgeit op 02-06-2023 13:17 ]

[Heaget]

Bedankt beide, het was niet direct de oplossing maar het heeft me wel anders laten kijken en de oplossing laten vinden. Er waren meerdere dingen verkeerd, ik gebruikte snake_case in plaats van camelCase, ik probeerde de functie aan te roepen met def ervoor en ik was het argument vergeten.

Soms vraag ik me weleens af hoe ik überhaupt af en toe werkende code schrijf..

Erg bedankt in ieder geval! Ik kan weer verder!