#!/usr/bin/env python3
# SPDX-License-Identifier: LGPL-2.1+
# systemd-networkd tests

import argparse
import os
import re
import shutil
import signal
import subprocess
import sys
import time
import unittest
from shutil import copytree

network_unit_file_path='/run/systemd/network'
networkd_runtime_directory='/run/systemd/netif'
networkd_ci_path='/run/networkd-ci'
network_sysctl_ipv6_path='/proc/sys/net/ipv6/conf'
network_sysctl_ipv4_path='/proc/sys/net/ipv4/conf'

dnsmasq_pid_file='/run/networkd-ci/test-test-dnsmasq.pid'
dnsmasq_log_file='/run/networkd-ci/test-dnsmasq-log-file'

systemd_lib_paths=['/usr/lib/systemd', '/lib/systemd']
which_paths=':'.join(systemd_lib_paths + os.getenv('PATH', os.defpath).lstrip(':').split(':'))

networkd_bin=shutil.which('systemd-networkd', path=which_paths)
resolved_bin=shutil.which('systemd-resolved', path=which_paths)
wait_online_bin=shutil.which('systemd-networkd-wait-online', path=which_paths)
networkctl_bin=shutil.which('networkctl', path=which_paths)
resolvectl_bin=shutil.which('resolvectl', path=which_paths)
timedatectl_bin=shutil.which('timedatectl', path=which_paths)

use_valgrind=False
enable_debug=True
env = {}
asan_options=None
lsan_options=None
ubsan_options=None

running_units = []

def check_output(*command, **kwargs):
    # This replaces both check_output and check_call (output can be ignored)
    command = command[0].split() + list(command[1:])
    return subprocess.check_output(command, universal_newlines=True, **kwargs).rstrip()

def call(*command, **kwargs):
    command = command[0].split() + list(command[1:])
    return subprocess.call(command, universal_newlines=True, **kwargs)

def run(*command, **kwargs):
    command = command[0].split() + list(command[1:])
    return subprocess.run(command, universal_newlines=True, **kwargs)

def is_module_available(module_name):
    lsmod_output = check_output('lsmod')
    module_re = re.compile(rf'^{re.escape(module_name)}\b', re.MULTILINE)
    return module_re.search(lsmod_output) or not call('modprobe', module_name, stderr=subprocess.DEVNULL)

def expectedFailureIfModuleIsNotAvailable(module_name):
    def f(func):
        if not is_module_available(module_name):
            return unittest.expectedFailure(func)
        return func

    return f

def expectedFailureIfERSPANModuleIsNotAvailable():
    def f(func):
        rc = call('ip link add dev erspan99 type erspan seq key 30 local 192.168.1.4 remote 192.168.1.1 erspan_ver 1 erspan 123', stderr=subprocess.DEVNULL)
        if rc == 0:
            call('ip link del erspan99')
            return func
        else:
            return unittest.expectedFailure(func)

    return f

def expectedFailureIfRoutingPolicyPortRangeIsNotAvailable():
    def f(func):
        rc = call('ip rule add from 192.168.100.19 sport 1123-1150 dport 3224-3290 table 7', stderr=subprocess.DEVNULL)
        if rc == 0:
            call('ip rule del from 192.168.100.19 sport 1123-1150 dport 3224-3290 table 7')
            return func
        else:
            return unittest.expectedFailure(func)

    return f

def expectedFailureIfRoutingPolicyIPProtoIsNotAvailable():
    def f(func):
        rc = call('ip rule add not from 192.168.100.19 ipproto tcp table 7', stderr=subprocess.DEVNULL)
        if rc == 0:
            call('ip rule del not from 192.168.100.19 ipproto tcp table 7')
            return func
        else:
            return unittest.expectedFailure(func)

    return f

def expectedFailureIfLinkFileFieldIsNotSet():
    def f(func):
        support = False
        rc = call('ip link add name dummy99 type dummy', stderr=subprocess.DEVNULL)
        if rc == 0:
            ret = run('udevadm info -w10s /sys/class/net/dummy99', stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
            if ret.returncode == 0 and 'E: ID_NET_LINK_FILE=' in ret.stdout.rstrip():
                support = True
            call('ip link del dummy99')

        if support:
         
   return func
        else:
            return unittest.expectedFailure(func)

    return f

def expectedFailureIfNexthopIsNotAvailable():
    def f(func):
        rc = call('ip nexthop list', stderr=subprocess.DEVNULL)
        if rc == 0:
            return func
        else:
            return unittest.expectedFailure(func)

    return f

def expectedFailureIfAlternativeNameIsNotAvailable():
    d
ef f(func):
        call('ip link add dummy98 type dummy', stderr=subprocess.DEVNULL)
        rc = call('ip link prop add dev dummy98 altname hogehogehogehogehoge', stderr=subprocess.DEVNULL)
        if rc == 0:
            return func
        else:
            return unittest.expectedFailure(func)

    return f

def setUpModule():
    global running_units

    os.makedirs(network_unit_file_path, exist_ok=True)
    os.makedirs(networkd_ci_path, exist_ok=True)

    shutil.rmtree(networkd_ci_path)
    copytree(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'conf'), networkd_ci_path)

    for u in ['systemd-networkd.socket', 'systemd-networkd.service', 'systemd-resolved.service', 'firewalld.service']:
        if call(f'systemctl is-active --quiet {u}') == 0:
            check_output(f'systemctl stop {u}')
            running_units.append(u)

    drop_in = [
        '[Unit]',
        'StartLimitIntervalSec=0',
        '[Service]',
        'Restart=no',
        'ExecStart=',
    ]
    if use_valgrind:
        drop_in += [
            'ExecStart=!!valgrind --track-origins=yes --leak-check=full --show-leak-kinds=all ' + networkd_bin,
            'PrivateTmp=yes'
        ]
    else:
        drop_in += ['ExecStart=!!' + networkd_bin]
    if enable_debug:
        drop_in += ['Environment=SYSTEMD_LOG_LEVEL=debug']
    if asan_options:
        drop_in += ['Environment=ASAN_OPTIONS="' + asan_options + '"']
    if lsan_options:
        drop_in += ['Environment=LSAN_OPTIONS="' + lsan_options + '"']
    if ubsan_options:
        drop_in += ['Environment=UBSAN_OPTIONS="' + ubsan_options + '"']
    if asan_options or lsan_options or ubsan_options:
        drop_in += ['SystemCallFilter=']
    if use_valgrind or asan_options or lsan_options or ubsan_options:
        drop_in += ['MemoryDenyWriteExecute=no']

    os.makedirs('/run/systemd/system/systemd-networkd.service.d', exist_ok=True)
    with open('/run/systemd/system/systemd-networkd.service.d/00-override.conf', mode='w') as f:
        f.write('\n'.join(drop_in))

    drop_in = [
        '[Service]',
        'Restart=no',
        'ExecStart=',
    ]
    if use_valgrind:
        drop_in += ['ExecStart=!!valgrind --track-origins=yes --leak-check=full --show-leak-kinds=all ' + resolved_bin]
    else:
        drop_in += ['ExecStart=!!' + resolved_bin]
    if enable_debug:
        drop_in += ['Environment=SYSTEMD_LOG_LEVEL=debug']
    if asan_options:
        drop_in += ['Environment=ASAN_OPTIONS="' + asan_options + '"']
    if lsan_options:
        drop_in += ['Environment=LSAN_OPTIONS="' + lsan_options + '"']
    if ubsan_options:
        drop_in += ['Environment=UBSAN_OPTIONS="' + ubsan_options + '"']
    if asan_options or lsan_options or ubsan_options:
        drop_in += ['SystemCallFilter=']
    if use_valgrind or asan_options or lsan_options or ubsan_options:
        drop_in += ['MemoryDenyWriteExecute=no']

    os.makedirs('/run/systemd/system/systemd-resolved.service.d', exist_ok=True)
    with open('/run/systemd/system/systemd-resolved.service.d/00-override.conf', mode='w') as f:
        f.write('\n'.join(drop_in))

    check_output('systemctl daemon-reload')
    print(check_output('systemctl cat systemd-networkd.service'))
    print(check_output('systemctl cat systemd-resolved.service'))
    check_output('systemctl restart systemd-resolved')

def tearDownModule():
    global running_units

    shutil.rmtree(networkd_ci_path)

    for u in ['systemd-networkd.service', 'systemd-resolved.service']:
        check_output(f'systemctl stop {u}')

    shutil.rmtree('/run/systemd/system/systemd-networkd.service.d')
    shutil.rmtree('/run/systemd/system/systemd-resolved.service.d')
    check_output('systemctl daemon-reload')

    for u in running_units:
        check_output(f'systemctl start {u}')

def read_l