#!/usr/bin/env python3

# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Linux specific tests."""

from __future__ import division

import collections
import contextlib
import errno
import glob
import io
import os
import re
import shutil
import socket
import struct
import textwrap
import time
import unittest
import warnings

import psutil
from psutil import LINUX
from psutil._compat import PY3
from psutil._compat import FileNotFoundError
from psutil._compat import basestring
from psutil._compat import u
from psutil.tests import GITHUB_ACTIONS
from psutil.tests import GLOBAL_TIMEOUT
from psutil.tests import HAS_BATTERY
from psutil.tests import HAS_CPU_FREQ
from psutil.tests import HAS_GETLOADAVG
from psutil.tests import HAS_RLIMIT
from psutil.tests import PYPY
from psutil.tests import TOLERANCE_DISK_USAGE
from psutil.tests import TOLERANCE_SYS_MEM
from psutil.tests import PsutilTestCase
from psutil.tests import ThreadTask
from psutil.tests import call_until
from psutil.tests import mock
from psutil.tests import reload_module
from psutil.tests import retry_on_failure
from psutil.tests import safe_rmpath
from psutil.tests import sh
from psutil.tests import skip_on_not_implemented
from psutil.tests import which


if LINUX:
    from psutil._pslinux import CLOCK_TICKS
    from psutil._pslinux import RootFsDeviceFinder
    from psutil._pslinux import calculate_avail_vmem
    from psutil._pslinux import open_binary


HERE = os.path.abspath(os.path.dirname(__file__))
SIOCGIFADDR = 0x8915
SIOCGIFCONF = 0x8912
SIOCGIFHWADDR = 0x8927
SIOCGIFNETMASK = 0x891b
SIOCGIFBRDADDR = 0x8919
if LINUX:
    SECTOR_SIZE = 512
EMPTY_TEMPERATURES = not glob.glob('/sys/class/hwmon/hwmon*')


# =====================================================================
# --- utils
# =====================================================================


def get_ipv4_address(ifname):
    import fcntl
    ifname = ifname[:15]
    if PY3:
        ifname = bytes(ifname, 'ascii')
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    with contextlib.closing(s):
        return socket.inet_ntoa(
            fcntl.ioctl(s.fileno(),
                        SIOCGIFADDR,
                        struct.pack('256s', ifname))[20:24])


def get_ipv4_netmask(ifname):
    import fcntl
    ifname = ifname[:15]
    if PY3:
        ifname = bytes(ifname, 'ascii')
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    with contextlib.closing(s):
        return socket.inet_ntoa(
            fcntl.ioctl(s.fileno(),
                        SIOCGIFNETMASK,
                        struct.pack('256s', ifname))[20:24])


def get_ipv4_broadcast(ifname):
    import fcntl
    ifname = ifname[:15]
    if PY3:
        ifname = bytes(ifname, 'ascii')
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    with contextlib.closing(s):
        return socket.inet_ntoa(
            fcntl.ioctl(s.fileno(),
                        SIOCGIFBRDADDR,
                        struct.pack('256s', ifname))[20:24])


def get_ipv6_addresses(ifname):
    with open("/proc/net/if_inet6", 'rt') as f:
        all_fields = []
        for line in f.readlines():
            fields = line.split()
            if fields[-1] == ifname:
                all_fields.append(fields)

        if len(all_fields) == 0:
            raise ValueError("could not find interface %r" % ifname)

    for i in range(0, len(all_fields)):
        unformatted = all_fields[i][0]
        groups = []
        for j in range(0, len(unformatted), 4):
            groups.append(unformatted[j:j + 4])
        formatted = ":".join(groups)
        packed = socket.inet_pton(socket.AF_INET6, formatted)
        all_fields[i] = socket.inet_ntop(socket.AF_INET6, packed)
    return all_fields


def get_mac_address(ifname):
    import fcntl
    ifname = ifname[:15]
    if PY3:
        ifname = bytes(ifname, 'ascii')
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    with contextlib.closing(s):
        info = fcntl.ioctl(
            s.fileno(), SIOCGIFHWADDR, struct.pack('256s', ifname))
        if PY3:
            def ord(x):
                return x
        else:
            import __builtin__
            ord = __builtin__.ord
        return ''.join(['%02x:' % ord(char) for char in info[18:24]])[:-1]


def free_swap():
    """Parse 'free' cmd and return swap memory's s total, used and free
    values.
    """
    out = sh(["free", "-b"], env={"LANG": "C.UTF-8"})
    lines = out.split('\n')
    for line in lines:
        if line.startswith('Swap'):
            _, total, used, free = line.split()
            nt = collections.namedtuple('free', 'total used free')
            return nt(int(total), int(used), int(free))
    raise ValueError(
        "can't find 'Swap' in 'free' output:\n%s" % '\n'.join(lines))


def free_physmem():
    """Parse 'free' cmd and return physical memory's total, used
    and free values.
    """
    # Note: free can have 2 different formats, invalidating 'shared'
    # and 'cached' memory which may have different positions so we
    # do not return them.
    # https://github.com/giampaolo/psutil/issues/538#issuecomment-57059946
    out = sh(["free", "-b"], env={"LANG": "C.UTF-8"})
    lines = out.split('\n')
    for line in lines:
        if line.startswith('Mem'):
            total, used, free, shared = \
                [int(x) for x in line.split()[1:5]]
            nt = collections.namedtuple(
                'free', 'total used free shared output')
            return nt(total, used, free, shared, out)
    raise ValueError(
        "can't find 'Mem' in 'free' output:\n%s" % '\n'.join(lines))


def vmstat(stat):
    out = sh(["vmstat", "-s"], env={"LANG": "C.UTF-8"})
    for line in out.split("\n"):
        line = line.strip()
        if stat in line:
            return int(line.split(' ')[0])
    raise ValueError("can't find %r in 'vmstat' output" % stat)


def get_free_version_info():
    out = sh(["free", "-V"]).strip()
    if 'UNKNOWN' in out:
        raise unittest.SkipTest("can't determine free version")
    return tuple(map(int, out.split()[-1].split('.')))


@contextlib.contextmanager
def mock_open_content(for_path, content):
    """Mock open() builtin and forces it to return a certain `content`
    on read() if the path being opened matches `for_path`.
    """
    def open_mock(name, *args, **kwargs):
        if name == for_path:
            if PY3:
                if isinstance(content, basestring):
                    return io.StringIO(content)
                else:
                    return io.BytesIO(content)
            else:
                return io.BytesIO(content)
        else:
            return orig_open(name, *args, **kwargs)

    orig_open = open
    patch_point = 'builtins.open' if PY3 else '__builtin__.open'
    with mock.patch(patch_point, create=True, side_effect=open_mock) as m:
        yield m


@contextlib.contextmanager
def mock_open_exception(for_path, exc):
    """Mock open() builtin and raises `exc` if the path being opened
    matches `for_path`.
    """
    def open_mock(name, *args, **kwargs):
        if name == for_path:
            raise exc
        else:
            return orig_open(name, *args, **kwargs)

    orig_open = open
    patch_point = 'builtins.open' if PY3 else '__builtin__.open'
    with mock.patch(patch_point, create=True, side_effect=open_mock) as m:
        yield m


# =====================================================================
# --- system virtual memory
# =====================================================================


@unittest.skipIf(not LINUX, "LINUX only")
class TestSystemVirtualMemory(PsutilTestCase):

    def test_total(self):
        # free_value = free_physmem().total
        # psutil_value = psutil.virtual_memory().total
        # self.assertEqual(free_value, psutil_value)
        vmstat_value = vmstat('total memory') * 1024
        psutil_value = psutil.virtual_memory().total
        self.assertAlmostEqual(
            vmstat_value, psutil_value, delta=TOLERANCE_SYS_MEM)

    @retry_on_failure()
    def test_used(self):
        # Older versions of procps used slab memory to calculate used memory.
        # This got changed in:
        # https://gitlab.com/procps-ng/procps/commit/
        #     05d751c4f076a2f0118b914c5e51cfbb4762ad8e
        if get_free_version_info() < (3, 3, 12):
            raise self.skipTest("old free version")
        free = free_physmem()
        free_value = free.used
        psutil_value = psutil.virtual_memory().used
        self.assertAlmostEqual(
            free_value, psutil_value, delta=TOLERANCE_SYS_MEM,
            msg='%s %s \n%s' % (free_value, psutil_value, free.output))

    @retry_on_failure()
    def test_free(self):
        vmstat_value = vmstat('free memory') * 1024
        psutil_value = psutil.virtual_memory().free
        self.assertAlmostEqual(
            vmstat_value, psutil_value, delta=TOLERANCE_SYS_MEM)

    @retry_on_failure()
    def test_buffers(self):
        vmstat_value = vmstat('buffer memory') * 1024
        psutil_value = psutil.virtual_memory().buffers
        self.assertAlmostEqual(
            vmstat_value, psutil_value, delta=TOLERANCE_SYS_MEM)

    @retry_on_failure()
    def test_active(self):
        vmstat_value = vmstat('active memory') * 1024
        psutil_value = psutil.virtual_memory().active
        self.assertAlmostEqual(
            vmstat_value, psutil_value, delta=TOLERANCE_SYS_MEM)

    @retry_on_failure()
    def test_inactive(self):
        vmstat_value = vmstat('inactive memory') * 1024
        psutil_value = psutil.virtual_memory().inactive
        self.assertAlmostEqual(
            vmstat_value, psutil_value, delta=TOLERANCE_SYS_MEM)

    @retry_on_failure()
    def test_shared(self):
        free = free_physmem()
        free_value = free.shared
        if free_value == 0:
            raise unittest.SkipTest("free does not support 'shared' column")
        psutil_value = psutil.virtual_memory().shared
        self.assertAlmostEqual(
            free_value, psutil_value, delta=TOLERANCE_SYS_MEM,
            msg='%s %s \n%s' % (free_value, psutil_value, free.output))

    @retry_on_failure()
    def test_available(self):
        # "free" output format has changed at some point:
        # https://github.com/giampaolo/psutil/issues/538#issuecomment-147192098
        out = sh(["free", "-b"])
        lines = out.split('\n')
        if 'available' not in lines[0]:
            raise unittest.SkipTest("free does not support 'available' column")
        else:
            free_value = int(lines[1].split()[-1])
            psutil_value = psutil.virtual_memory().available
            self.assertAlmostEqual(
                free_value, psutil_value, delta=TOLERANCE_SYS_MEM,
                msg='%s %s \n%s' % (free_value, psutil_value, out))

    def test_warnings_on_misses(self):
        # Emulate a case where /proc/meminfo provides few info.
        # psutil is supposed to set the missing fields to 0 and
        # raise a warning.
        with mock_open_content(
            '/proc/meminfo',
            textwrap.dedent("""\
                Active(anon):    6145416 kB
                Active(file):    2950064 kB
                Inactive(anon):   574764 kB
                Inactive(file):  1567648 kB
                MemAvailable:         -1 kB
                MemFree:         2057400 kB