# SPDX-FileCopyrightText: All Contributors to the PyTango project
# SPDX-License-Identifier: LGPL-3.0-or-later
"""Client tests that run against the standard TangoTest device.

Due to a TANGO 9 issue (#821), the device is run without any database.
Note that this means that various features won't work:

 * No device configuration via properties.
 * No event generated by the server.
 * No memorized attributes.
 * No device attribute configuration via the database.

So don't even try to test anything of the above as it will not work
and is even likely to crash the device (!)

"""

import asyncio
import gc
import multiprocessing
import time
import weakref

import pytest

from tango import (
    AttributeInfo,
    AttributeInfoEx,
    AttributeInfoList,
    AttributeInfoListEx,
    DeviceInfo,
    DeviceProxy,
    EventType,
    ExtractAs,
    GreenMode,
    Group,
    PyTangoUserWarning,
    constants,
)
from tango.constants import AllAttr
from tango.server import Device, attribute
from tango.test_utils import (
    DeviceTestContext,
    assert_close,
    bytes_devstring,
    convert_to_type,
    str_devstring,
)
from tango.utils import (
    EventCallback,
    is_str_type,
    is_int_type,
    is_float_type,
    is_bool_type,
)

ATTRIBUTES = [
    "State",
    "Status",
    "ampli",
    "boolean_image",
    "boolean_image_ro",
    "boolean_scalar",
    "boolean_spectrum",
    "boolean_spectrum_ro",
    "double_image",
    "double_image_ro",
    "double_scalar",
    "double_scalar_rww",
    "double_scalar_w",
    "double_spectrum",
    "double_spectrum_ro",
    "echo_mode",
    "enum_image",
    "enum_image_ro",
    "enum_scalar",
    "enum_scalar_ro",
    "enum_spectrum",
    "enum_spectrum_ro",
    "float_image",
    "float_image_ro",
    "float_scalar",
    "float_spectrum",
    "float_spectrum_ro",
    "freq",
    "long64_image_ro",
    "long64_scalar",
    "long64_spectrum_ro",
    "long_image",
    "long_image_ro",
    "long_scalar",
    "long_scalar_rww",
    "long_scalar_w",
    "long_spectrum",
    "long_spectrum_ro",
    "no_value",
    "short_image",
    "short_image_ro",
    "short_scalar",
    "short_scalar_ro",
    "short_scalar_rww",
    "short_scalar_w",
    "short_spectrum",
    "short_spectrum_ro",
    "string_image",
    "string_image_ro",
    "string_scalar",
    "string_spectrum",
    "string_spectrum_ro",
    "throw_exception",
    "uchar_image",
    "uchar_image_ro",
    "uchar_scalar",
    "uchar_spectrum",
    "uchar_spectrum_ro",
    "ulong64_image_ro",
    "ulong64_scalar",
    "ulong64_spectrum_ro",
    "ulong_image_ro",
    "ulong_scalar",
    "ulong_spectrum_ro",
    "ushort_image",
    "ushort_image_ro",
    "ushort_scalar",
    "ushort_spectrum",
    "ushort_spectrum_ro",
    "wave",
]

WRITABLE_SCALAR_ATTRIBUTES = [
    a for a in ATTRIBUTES if "scalar" in a and a.split("_")[-1] not in ("ro", "rww")
]
WRITABLE_SPECTRUM_ATTRIBUTES = [
    a for a in ATTRIBUTES if "spectrum" in a and a.split("_")[-1] not in ("ro", "rww")
]

TEST_DOUBLE_ATTRIBUTES = (
    ("double_scalar_w", -28.2),
    ("double_spectrum", [-28.2, 23.4]),
    ("double_image", [[-28.2, 23.4], [-4.9, 6.5]]),
)

# Helpers


def ping_device(proxy):
    if proxy.get_green_mode() == GreenMode.Asyncio:
        asyncio.get_event_loop().run_until_complete(proxy.ping())
    else:
        proxy.ping()


# Fixtures


@pytest.fixture(params=ATTRIBUTES)
def attributes(request):
    return request.param


@pytest.fixture(
    params=[a for a in ATTRIBUTES if a not in ("no_value", "throw_exception")]
)
def readable_attribute(request):
    return request.param


@pytest.fixture(params=WRITABLE_SCALAR_ATTRIBUTES)
def writable_scalar_attribute(request):
    return request.param


@pytest.fixture(params=WRITABLE_SPECTRUM_ATTRIBUTES)
def writable_spectrum_attribute(request):
    return request.param


@pytest.fixture
def simple_device_fqdn():
    class TestDevice(Device):
        pass

    context = DeviceTestContext(TestDevice, host="127.0.0.1")
    context.start()
    yield context.get_device_access()
    context.stop()


# Tests


def test_ping(tango_test):
    duration = tango_test.ping(wait=True)
    assert isinstance(duration, int)


def test_info(tango_test):
    info = tango_test.info()
    assert isinstance(info, DeviceInfo)
    assert info.dev_class == "TangoTest"
    info_dict = info.version_info
    assert isinstance(info_dict, dict)
    assert len(info_dict) > 0


def test_read_attribute(tango_test, readable_attribute):
    "Check that readable attributes can be read"
    # For read-only string spectrum and read-only string image types,
    # the following error is very likely to be raised:
    # -> MARSHAL CORBA system exception: MARSHAL_PassEndOfMessage
    # An explicit sleep fixes the problem, but it's annoying to maintain

    if readable_attribute in ["string_image_ro", "string_spectrum_ro"]:
        pytest.xfail()
    if readable_attribute == 'echo_mode':
        pytest.skip("This test is failing and temporarily disabled.")
    tango_test.read_attribute(readable_attribute, wait=True)


def test_read_write_attribute_with_green_modes(tango_test_with_green_modes):
    """
    Check that attributes can be read/write with all green modes
    """
    for attr_name, write_value in TEST_DOUBLE_ATTRIBUTES:
        tango_test_with_green_modes.write_attribute(attr_name, write_value, wait=True)
        read_attr = tango_test_with_green_modes.read_attribute(attr_name, wait=True)
        assert_close(read_attr.value, write_value)
        assert_close(read_attr.w_value, write_value)


@pytest.mark.asyncio
async def test_high_level_api_for_asyncio(tango_test):
    tango_test.set_green_mode(GreenMode.Asyncio)
    _ = await tango_test.long_scalar
    _ = await getattr(tango_test, "long_scalar")
    _ = await tango_test["long_scalar"]


def test_write_scalar_attribute(tango_test, writable_scalar_attribute):
    "Check that writable scalar attributes can be written"
    attr_name = writable_scalar_attribute
    config = tango_test.get_attribute_config(attr_name, wait=True)
    if is_bool_type(config.data_type):
        tango_test.write_attribute(attr_name, True, wait=True)
    elif is_int_type(config.data_type):
        tango_test.write_attribute(attr_name, 76, wait=True)
    elif is_float_type(config.data_type):
        tango_test.write_attribute(attr_name, -28.2, wait=True)
    elif is_str_type(config.data_type):
        tango_test.write_attribute(attr_name, "hello", wait=True)
    else:
        pytest.xfail("Not currently testing this type")


def test_write_read_spectrum_attribute(
    tango_test, writable_spectrum_attribute, extract_as
):
    "Check that writable spectrum attributes can be written and read"
    requested_type, expected_type = extract_as
    attr_name = writable_spectrum_attribute
    config = tango_test.get_attribute_config(attr_name, wait=True)
    if is_bool_type(config.data_type):
        write_values = [True, False]
    elif is_int_type(config.data_type):
        write_values = [76, 77]
    elif is_float_type(config.data_type):
        if requested_type == ExtractAs.String:
            # it is hard to find a proper float values, which can be converted to str,
            # most of them case UnicodeDecodeError, so we use the most simple one
            write_values = [0, 0]
        else:
            write_values = [-28.2, 44.3]
    elif is_str_type(config.data_type):
        if requested_type == ExtractAs.Numpy:
            expected_type = tuple
        if requested_type in [ExtractAs.ByteArray, ExtractAs.Bytes, ExtractAs.String]:
            pytest.xfail(
                "Conversion from (str,) to ByteArray, Bytes and String not supported. May be fixed in future"
            )
        write_values = ["hello", "hola"]
    else:
        pytest.xfail("Not currently testing this type")

    tango_test.write_attribute(attr_name, write_values, wait=True)
    read_attr = tango_test.read_attribute(attr_name, extract_as=requested_type)

    assert isinstance(read_attr.value, expected_type)
    assert_close(
        read_attr.value, convert_to_type(write_values, config.data_type, expected_type)
    )

    assert isinstance(read_attr.w_value, expected_type)
    assert_close(
        read_attr.w_value,
        convert_to_type(write_values, config.data_type, expected_type),
    )


def test_write_read_empty_spectrum_attribute(tango_test, writable_spectrum_attribute):
    "Check that writing empty list to spectrum attribute reads back as None."
    attr_name = writable_spectrum_attribute
    config = tango_test.get_attribute_config(attr_name, wait=True)
    if is_str_type(config.data_type):
        pytest.xfail(
            "Conversion from (str,) to numpy not supported. Probably, may be fixed in future"
        )

    tango_test.write_attribute(attr_name, [], wait=True)
    read_attr = tango_test.read_attribute(attr_name, wait=True)
    assert not len(read_attr.value)


def test_write_read_string_attribute(tango_test):
    attr_name = "string_scalar"
    bytes_big = 100000 * b"big data "
    str_big = bytes_big.decode("latin-1")

    values = [
        b"",
        "",
        "Hello, World!",
        b"Hello, World!",
        bytes_devstring,
        str_devstring,
        bytes_big,
        str_big,
    ]

    expected_values = [
        "",
        "",
        "Hello, World!",
        "Hello, World!",
        str_devstring,
        str_devstring,
        str_big,
        str_big,
    ]

    for value, expected_value in zip(values, expected_values):
        tango_test.write_attribute(attr_name, value, wait=True)
        result = tango_test.read_attribute(attr_name, wait=True)
        assert result.value == expected_value

    attr_name = "string_spectrum"
    for value, expected_value in zip(values, expected_values):
        tango_test.write_attribute(attr_name, ["", value, ""], wait=True)
        result = tango_test.read_attribute(attr_name, wait=True)
        assert result.value[1] == expected_value

    attr_name = "string_image"
    for value, expected_value in zip(values, expected_values):
        tango_test.write_attribute(attr_name, [[value], [value]], wait=True)
        result = tango_test.read_attribute(attr_name, wait=True)
        assert result.value == ((expected_value,), (expected_value,))


def test_set_non_existent_attribute_raises_by_default(tango_test):
    with pytest.raises(AttributeError, match="some_invalid_name"):
        tango_test.some_invalid_name = "123"


def test_set_non_existent_attribute_allowed_if_dynamic_interface_unfrozen(tango_test):
    with pytest.warns(PyTangoUserWarning):
        tango_test.unfreeze_dynamic_interface()
    tango_test.some_invalid_name = "123"
    assert tango_test.some_invalid_name == "123"


def test_dynamic_interface_can_be_toggled(tango_test):
    with pytest.warns(PyTangoUserWarning):
        tango_test.unfreeze_dynamic_interface()
    tango_test.some_invalid_name = "456"
    assert tango_test.some_invalid_name == "456"
    tango_test.freeze_dynamic_interface()
    with pytest.raises(AttributeError, match="another_invalid_name"):
        tango_test.another_invalid_name = "123"


def test_dynamic_interface_flag_can_be_read(tango_test):
    with pytest.warns(PyTangoUserWarning):
        tango_test.unfreeze_dynamic_interface()
    assert not tango_test.is_dynamic_interface_frozen()
    tango_test.freeze_dynamic_interface()
    assert tango_test.is_dynamic_interface_frozen()


def test_dynamic_interface_only_applies_to_device_proxy_instance(tango_test):
    other_proxy = DeviceProxy(tango_test.adm_name())
    with pytest.warns(PyTangoUserWarning):
        tango_test.unfreeze_dynamic_interface()
    other_proxy.freeze_dynamic_interface()
    assert not tango_test.is_dynamic_interface_frozen()
    assert other_proxy.is_dynamic_interface_frozen()


def test_dynamic_interface_unfreeze_generates_a_user_warning(tango_test):
    with pytest.warns(PyTangoUserWarning):
        tango_test.unfreeze_dynamic_interface()


@pytest.mark.skip(reason="This test is failing and temporarily disabled.")
def test_read_attribute_config(tango_test):
    all_conf = tango_test.get_attribute_config(ATTRIBUTES)
    assert len(all_conf) == len(ATTRIBUTES)
    assert set([conf.name for conf in all_conf]) == set(ATTRIBUTES)

    for attr in ATTRIBUTES:
        tango_test.get_attribute_config(attr)


def test_read_attribute_config_ex():
    class TestDevice(Device):

        @attribute(dtype=int, unit="mA")
        def attr_config_int(self):
            return 1

        @attribute(dtype=bool)
        def attr_config_bool(self):
            return True

    def assert_attr_config_ok(dev_proxy):
        # testing that call to get_attribute_config_ex for all types of
        # input arguments gives same result and doesn't raise an exception
        ac1 = dev_proxy.get_attribute_config_ex("attr_config_int")
        ac2 = dev_proxy.get_attribute_config_ex(["attr_config_int"])
        assert repr(ac1) == repr(ac2)

    def assert_multiple_attrs_config_ok(dev_proxy):
        # testing that querying multiple attributes gives same result and
        # doesn't raise an exception
        ac1 = dev_proxy.get_attribute_config_ex("attr_config_int")
        ac2 = dev_proxy.get_attribute_config_ex("attr_config_bool")
        acs = dev_proxy.get_attribute_config_ex(("attr_config_int", "attr_config_bool"))
        acs_2 = dev_proxy.get_attribute_config_ex(
            ["attr_config_int", "attr_config_bool"]
        )

        acs_all = dev_proxy.get_attribute_config_ex(AllAttr)

        assert repr(ac1[0]) == repr(acs[0]) == repr(acs_2[0]) == repr(
            acs_all[0]
        ) and repr(ac2[0]) == repr(acs[1]) == repr(acs_2[1]) == repr(acs_all[1])

    with DeviceTestContext(TestDevice) as proxy:
        assert_attr_config_ok(proxy)
        assert_multiple_attrs_config_ok(proxy)

@pytest.mark.skip(reason="This test is failing and temporarily disabled.")
def test_attribute_list_query(tango_test):
    attrs = tango_test.attribute_list_query()
    assert isinstance(attrs, AttributeInfoList)
    assert all(isinstance(a, AttributeInfo) for a in attrs)
    assert {a.name for a in attrs} == set(ATTRIBUTES)


@pytest.mark.skip(reason="This test is failing and temporarily disabled.")
def test_attribute_list_query_ex(tango_test):
    attrs = tango_test.attribute_list_query_ex()
    assert isinstance(attrs, AttributeInfoListEx)
    assert all(isinstance(a, AttributeInfoEx) for a in attrs)
    assert {a.name for a in attrs} == set(ATTRIBUTES)


def test_device_proxy_dir_method(tango_test):
    lst = dir(tango_test)
    attrs = tango_test.get_attribute_list()
    cmds = tango_test.get_command_list()
    with pytest.warns(DeprecationWarning):
        pipes = tango_test.get_pipe_list()
    methods = dir(type(tango_test))
    internals = tango_test.__dict__.keys()
    # Check attributes
    assert set(attrs) < set(lst)
    assert set(map(str.lower, attrs)) < set(lst)
    # Check commands
    assert set(cmds) < set(lst)
    assert set(map(str.lower, cmds)) < set(lst)
    # Check pipes
    assert set(pipes) < set(lst)
    assert set(map(str.lower, pipes)) < set(lst)
    # Check internals
    assert set(methods) <= set(lst)
    # Check internals
    assert set(internals) <= set(lst)


def test_device_polling_command(tango_test):
    dct = {"SwitchStates": 1000, "DevVoid": 10000, "DumpExecutionState": 5000}

    for command, period in dct.items():
        tango_test.poll_command(command, period)

    ans = tango_test.polling_status()
    for info in ans:
        lines = info.split("\n")
        command = lines[0].split("= ")[1]
        period = int(lines[1].split("= ")[1])
        assert dct[command] == period


def test_device_polling_attribute(tango_test):
    dct = {"boolean_scalar": 1000, "double_scalar": 10000, "long_scalar": 5000}

    for attr, poll_period in dct.items():
        tango_test.poll_attribute(attr, poll_period)

    ans = tango_test.polling_status()
    for x in ans:
        lines = x.split("\n")
        attr = lines[0].split("= ")[1]
        poll_period = int(lines[1].split("= ")[1])
        assert dct[attr] == poll_period


def test_command_string(tango_test):
    cmd_name = "DevString"
    bytes_big = 100000 * b"big data "
    str_big = bytes_big.decode("latin-1")

    values = [
        b"",
        "",
        "Hello, World!",
        b"Hello, World!",
        bytes_devstring,
        str_devstring,
        bytes_big,
        str_big,
    ]

    expected_values = [
        "",
        "",
        "Hello, World!",
        "Hello, World!",
        str_devstring,
        str_devstring,
        str_big,
        str_big,
    ]

    for value, expected_value in zip(values, expected_values):
        result = tango_test.command_inout(cmd_name, value, wait=True)
        assert result == expected_value

    cmd_name = "DevVarStringArray"
    for value, expected_value in zip(values, expected_values):
        result = tango_test.command_inout(cmd_name, [value, value], wait=True)
        assert result == [expected_value, expected_value]

    cmd_name = "DevVarLongStringArray"
    for value, expected_value in zip(values, expected_values):
        result = tango_test.command_inout(
            cmd_name, [[-10, 200], [value, value]], wait=True
        )
        assert len(result) == 2
        assert_close(result[0], [-10, 200])
        assert_close(result[1], [expected_value, expected_value])


def test_command_raises_type_error_for_bad_input(tango_test):
    expected_message = "Invalid input argument for command"
    with pytest.raises(TypeError, match=expected_message):
        tango_test.command_inout("DevDouble", "123.0")
    with pytest.raises(TypeError, match=expected_message):
        tango_test.command_inout("DevLong64", [1, 2, 3])
    with pytest.raises(TypeError, match=expected_message):
        tango_test.command_inout("DevString", [{"invalid": "type"}, 123.0])
    with pytest.raises(TypeError, match=expected_message):
        tango_test.command_inout("DevVarStringArray", [1, 2, 3])
    with pytest.raises(TypeError, match=expected_message):
        tango_test.command_inout("DevVarLong64Array", ["1", 2.0, {3}])


def test_repr_uses_info(green_mode_device_proxy, simple_device_fqdn):
    proxy = green_mode_device_proxy(simple_device_fqdn)
    assert repr(proxy) == "TestDevice(test/nodb/testdevice)"


def test_repr_default_if_info_unavailable(green_mode_device_proxy, simple_device_fqdn):
    proxy = green_mode_device_proxy(simple_device_fqdn)

    def bad_info(self):
        raise RuntimeError("Break info for test")

    proxy.__class__.info = bad_info

    assert repr(proxy) == "Device(test/nodb/testdevice)"


def test_multiple_repr_calls_only_call_info_once(
    green_mode_device_proxy, simple_device_fqdn
):
    proxy = green_mode_device_proxy(simple_device_fqdn)

    def mock_info(self):
        self.info_call_count += 1
        return self.info_orig()

    proxy.__class__.info_orig = proxy.info
    proxy.__class__.info = mock_info
    proxy.__dict__["info_call_count"] = 0

    repr(proxy)
    assert proxy.info_call_count == 1
    repr(proxy)
    assert proxy.info_call_count == 1


def test_no_memory_leak_for_repr(green_mode_device_proxy, simple_device_fqdn):
    proxy = green_mode_device_proxy(simple_device_fqdn)
    ping_device(proxy)
    weak_ref = weakref.ref(proxy)

    repr(proxy)

    # clear strong reference and check if object can be garbage collected
    del proxy
    assert_object_released_after_gc(weak_ref)


def test_no_memory_leak_for_str(green_mode_device_proxy, simple_device_fqdn):
    proxy = green_mode_device_proxy(simple_device_fqdn)
    ping_device(proxy)
    weak_ref = weakref.ref(proxy)

    str(proxy)

    # clear strong reference and check if object can be garbage collected
    del proxy
    assert_object_released_after_gc(weak_ref)


def test_no_cyclic_ref_for_proxy(green_mode_device_proxy, simple_device_fqdn):
    proxy = green_mode_device_proxy(simple_device_fqdn)
    ping_device(proxy)
    weak_ref = weakref.ref(proxy)

    # clear strong reference and check if object is immediately released
    del proxy
    assert_object_released_without_gc(weak_ref)


def group_client_lifecycle(device_trl):
    group = Group("test")
    group_weak_ref = weakref.ref(group)
    group.add(device_trl)
    del group
    assert_object_released_without_gc(group_weak_ref)


def device_proxy_lifecycle(device_trl):
    proxy = DeviceProxy(device_trl)
    proxy_weak_ref = weakref.ref(proxy)
    del proxy
    assert_object_released_without_gc(proxy_weak_ref)


@pytest.mark.extra_src_test
@pytest.mark.parametrize("subject", [group_client_lifecycle, device_proxy_lifecycle])
def test_client_destructor_does_not_deadlock(
    tango_test_process_device_trl_with_function_scope, subject
):
    proc, device_trl = tango_test_process_device_trl_with_function_scope

    # Create client to be run in a subprocess.
    # We want this client to wait for a "not connected event"
    # while rapidly creating and destroying objects (using "subject" function)
    # to try to induce a race condition.
    # The "not connected event" is about 2 seconds after the event heartbeat period
    # of 10 seconds - we add some margin either side.
    # We use some Event objects for additional synchronisation
    heartbeat_margin = 0.5
    delay_before_heartbeat = constants.EVENT_HEARTBEAT_PERIOD - heartbeat_margin
    max_time_until_not_connected_event = delay_before_heartbeat + 3.0
    process_startup_timeout = 1.0
    process_exit_timeout = 1.0
    proxy_created = multiprocessing.Event()
    process_terminated = multiprocessing.Event()
    client = multiprocessing.Process(
        target=continuously_call_subject_during_not_connected_event,
        args=(
            device_trl,
            proxy_created,
            process_terminated,
            delay_before_heartbeat,
            max_time_until_not_connected_event,
            subject,
        ),
    )
    # start client, and wait for it to create a DeviceProxy
    client.start()
    assert proxy_created.wait(timeout=process_startup_timeout)

    # now that client has created DeviceProxy, we terminate the device server process
    # and inform the client
    proc.terminate()
    process_terminated.set()

    # finally, wait for client.
    # - if deadlocked, the exitcode will be None.
    # - if it has some other problem, the exitcode will be non-zero.
    client.join(timeout=max_time_until_not_connected_event + process_exit_timeout)
    assert client.exitcode == 0


def continuously_call_subject_during_not_connected_event(
    device_trl: str,
    proxy_created: multiprocessing.Event,
    process_terminated: multiprocessing.Event,
    delay_before_heartbeat: float,
    max_duration: float,
    subject: callable,
):
    # stateless subscription
    cb = EventCallback()
    proxy = DeviceProxy(device_trl)
    proxy_created.set()
    eid = proxy.subscribe_event(
        "boolean_scalar", EventType.CHANGE_EVENT, cb, stateless=True
    )
    assert eid > 0

    # wait for Tango device server to be terminated by parent
    terminate_timeout = 1.0
    assert process_terminated.wait(timeout=terminate_timeout)

    # wait for a "not connected event" while rapidly calling the
    # subject function (which will create and destroy objects)
    # to try to induce the race condition
    start_time = time.time()
    time.sleep(delay_before_heartbeat)
    while time.time() - start_time < max_duration:
        subject(device_trl)


def assert_object_released_after_gc(weak_ref):
    gc.collect()
    assert weak_ref() is None


def assert_object_released_without_gc(weak_ref):
    try:
        assert weak_ref() is None
    except AssertionError:
        if weak_ref().get_green_mode() == GreenMode.Asyncio:
            pytest.xfail("Sometimes fails with a concurrent.Future ref cycle")
        else:
            raise