test/imports.uts

% Import tests
~ not_pypy

+ Import tests
~ imports

= Prepare importing all scapy files

import os
import glob
import subprocess
import re
import time
import sys
from scapy.consts import WINDOWS, OPENBSD

# DEV: to add your file to this list, make sure you have
# a GREAT reason.
EXCEPTIONS = [
    "scapy.__main__",
    "scapy.all",
    "scapy.contrib.automotive*",
    "scapy.contrib.cansocket*",
    "scapy.contrib.isotp*",
    "scapy.contrib.scada*",
    "scapy.layers.all",
    "scapy.main",
]

if WINDOWS:
    EXCEPTIONS.append("scapy.layers.tuntap")

EXCEPTION_PACKAGES = [
    "arch",
    "libs",
    "modules",
    "tools",
]

ALL_FILES = [
    "scapy." + re.match(".*scapy\\" + os.path.sep + "(.*)\\.py$", x).group(1).replace(os.path.sep, ".")
    for x in glob.iglob(scapy_path('/scapy/**/*.py'), recursive=True)
]
ALL_FILES = [
    x for x in ALL_FILES if
    not any(x == y if y[-1] != "*" else x.startswith(y[:-1]) for y in EXCEPTIONS) and
    x.split(".")[1] not in EXCEPTION_PACKAGES
]

NB_PROC = 1 if WINDOWS or OPENBSD else 4

def append_processes(processes, filename):
    processes.append(
        (subprocess.Popen(
            [sys.executable, "-c", "import %s" % filename],
            stderr=subprocess.PIPE, encoding="utf8"),
         time.time(),
         filename))

def check_processes(processes):
    for i, tup in enumerate(processes):
        proc, start_ts, file = tup
        errs = ""
        try:
            _, errs = proc.communicate(timeout=0.5)
        except subprocess.TimeoutExpired:
            if time.time() - start_ts > 30:
                proc.kill()
                errs = "Timed out (>30s)!"
        if proc.returncode is None:
            continue
        else:
            print("Finished %s with %d after %f sec" %
                  (file, proc.returncode, time.time() - start_ts))
            if proc.returncode != 0:
                for p in processes:
                    p[0].kill()
                raise Exception(
                    "Importing the file '%s' failed !\\n%s" % (file, errs))
            del processes[i]
            return


def import_all(FILES):
    processes = list()
    while len(processes) == NB_PROC:
        check_processes(processes)
    for filename in FILES:
        check_processes(processes)
        if len(processes) < NB_PROC:
            append_processes(processes, filename)


= Try importing all core separately

import_all(x for x in ALL_FILES if "layers" not in x and "contrib" not in x)

= Try importing all layers separately

import_all(x for x in ALL_FILES if "layers" in x)

= Try importing all contribs separately

import_all(x for x in ALL_FILES if "contrib" in x)