from textwrap import dedent
from unittest import TestCase

from pcs_test.tools import fixture
from pcs_test.tools.command_env import get_env_tools
from pcs_test.tools.custom_mock import patch_getaddrinfo

from pcs.common import report_codes
from pcs.lib.commands import cluster

class UpdateLink(TestCase):
    def setUp(self):
        self.env_assist, self.config = get_env_tools(self)
        self.config.env.set_known_nodes(["node1", "node2"])
        self.link_options = {"mcastport": "12345"}
        self.node_addr_map = {"node2": "node2-addr2a"}
        self.before = dedent("""\
            totem {
                transport: knet

                interface {
                    linknumber: 2
                    mcastport: 1234
                }
            }

            nodelist {
                node {
                    ring0_addr: node1-addr0
                    ring2_addr: node1-addr2
                    name: node1
                    nodeid: 1
                }

                node {
                    ring0_addr: node2-addr0
                    ring2_addr: node2-addr2
                    name: node2
                    nodeid: 2
                }
            }
            """
        )
        self.after = dedent("""\
            totem {
                transport: knet

                interface {
                    linknumber: 2
                    mcastport: 12345
                }
            }

            nodelist {
                node {
                    ring0_addr: node1-addr0
                    ring2_addr: node1-addr2
                    name: node1
                    nodeid: 1
                }

                node {
                    ring0_addr: node2-addr0
                    ring2_addr: node2-addr2a
                    name: node2
                    nodeid: 2
                }
            }
            """
        )

    def test_not_live(self):
        (self.config
            .env.set_corosync_conf_data("")
        )
        self.env_assist.assert_raise_library_error(
            lambda: cluster.update_link(
                self.env_assist.get_env(),
                "0",
                {},
                {}
            ),
            [
                fixture.error(
                    report_codes.LIVE_ENVIRONMENT_REQUIRED,
                    forbidden_options=["COROSYNC_CONF"]
                ),
            ],
            expected_in_processor=False
        )

    def test_missing_input_data(self):
        (self.config
            .corosync_conf.load_content(self.before)
            .env.push_corosync_conf(corosync_conf_text=self.before)
        )
        cluster.update_link(
            self.env_assist.get_env(),
            "0",
            {},
            {}
        )
        # Reports from pushing corosync.conf are produced in env. That code is
        # hidden in self.config.env.push_corosync_conf.
        self.env_assist.assert_reports([])

    def test_offline_nodes(self):
        patch_getaddrinfo(self, self.node_addr_map.values())
        (self.config
            .corosync_conf.load_content(self.before)
            .env.push_corosync_conf(
                corosync_conf_text=self.after,
                skip_offline_targets=True,
            )
        )
        cluster.update_link(
            self.env_assist.get_env(),
            "2",
            self.node_addr_map,
            self.link_options,
            force_flags=[report_codes.SKIP_OFFLINE_NODES]
        )

    def test_unresolvable_addresses(self):
        (self.config
            .corosync_conf.load_content(self.before)
        )
        self.env_assist.assert_raise_library_error(
            lambda: cluster.update_link(
                self.env_assist.get_env(),
                "2",
                self.node_addr_map,
                self.link_options
            ),
            []
        )
        self.env_assist.assert_reports(
            [
                fixture.error(
                    report_codes.NODE_ADDRESSES_UNRESOLVABLE,
                    force_code=report_codes.FORCE_NODE_ADDRESSES_UNRESOLVABLE,
                    address_list=list(self.node_addr_map.values()),
                )
            ]
        )

    def test_unresolvable_addresses_forced(self):
        (self.config
            .corosync_conf.load_content(self.before)
            .env.push_corosync_conf(corosync_conf_text=self.after)
        )
        cluster.update_link(
            self.env_assist.get_env(),
            "2",
            self.node_addr_map,
            self.link_options,
            force_flags=[report_codes.FORCE]
        )
        self.env_assist.assert_reports(
            [
                fixture.warn(
                    report_codes.NODE_ADDRESSES_UNRESOLVABLE,
                    address_list=list(self.node_addr_map.values()),
                )
            ]
        )

class UpdateLinkKnet(TestCase):
    def setUp(self):
        self.env_assist, self.config = get_env_tools(self)
        self.config.env.set_known_nodes(["node1", "node2"])
        self.before = dedent("""\
            totem {
                transport: knet

                interface {
                    linknumber: 2
                    mcastport: 1234
                    knet_transport: sctp
                }
            }

            nodelist {
                node {
                    ring0_addr: node1-addr0
                    ring2_addr: node1-addr2
                    name: node1
                    nodeid: 1
                }

                node {
                    ring0_addr: node2-addr0
                    ring2_addr: node2-addr2
                    name: node2
                    nodeid: 2
                }
            }
            """
        )

    def test_success(self):
        after = dedent("""\
            totem {
                transport: knet

                interface {
                    linknumber: 2
                    knet_transport: udp
                    knet_link_priority: 10
                }
            }

            nodelist {
                node {
                    ring0_addr: node1-addr0
                    ring2_addr: node1-addr2
                    name: node1
                    nodeid: 1
                }

                node {
                    ring0_addr: node2-addr0
                    ring2_addr: node2-addr2a
                    name: node2
                    nodeid: 2
                }
            }
            """
        )
        patch_getaddrinfo(self, ["node2-addr2a"])
        (self.config
            .corosync_conf.load_content(self.before)
            .env.push_corosync_conf(corosync_conf_text=after)
        )
        cluster.update_link(
            self.env_assist.get_env(),
            "2",
            {"node2": "node2-addr2a"},
            {"mcastport": "", "transport": "udp", "link_priority": "10"}
        )
        # Reports from pushing corosync.conf are produced in env. That code is
        # hidden in self.config.env.push_corosync_conf.
        self.env_assist.assert_reports([])

    def test_validation(self):
        patch_getaddrinfo(self, ["node2-addr0"])
        (self.config
            .corosync_conf.load_content(self.before)
        )
        self.env_assist.assert_raise_library_error(
            lambda: cluster.update_link(
                self.env_assist.get_env(),
                "2",
                {
                    "nodeX": "addr-new",
                    "node2": "",
                    "node1": "node2-addr0",
                },
                {
                    "wrong": "option",
                    "transport": "unknown",
                    "link_priority": 10,
                }
            ),
            []
        )
        self.env_assist.assert_reports(
            [
                fixture.error(
                    report_codes.INVALID_OPTION_VALUE,
                    option_value="unknown",
                    option_name="transport",
                    allowed_values=("sctp", "udp"),
                    cannot_be_empty=False,
                    forbidden_characters=None,
                ),
                fixture.error(
                    report_codes.INVALID_OPTIONS,
                    option_names=["wrong"],
                    option_type="link",
                    allowed=[
                        "link_priority", "mcastport", "ping_interval",
                        "ping_precision", "ping_timeout", "pong_count",
                        "transport",
                    ],
                    allowed_patterns=[],
                ),
                fixture.error(
                    report_codes.NODE_NOT_FOUND,
                    node="nodeX",
                    searched_types=[],
                ),
                fixture.error(
                    report_codes.NODE_ADDRESSES_CANNOT_BE_EMPTY,
                    node_name_list=["node2"],
                ),
                fixture.error(
                    report_codes.NODE_ADDRESSES_UNRESOLVABLE,
                    force_code=report_codes.FORCE_NODE_ADDRESSES_UNRESOLVABLE,
                    address_list=["addr-new"],
                ),
                fixture.error(
                    report_codes.NODE_ADDRESSES_ALREADY_EXIST,
                    address_list=["node2-addr0"],
                ),
            ]
        )

class UpdateLinkUdp(TestCase):
    def setUp(self):
        self.env_assist, self.config = get_env_tools(self)
        self.config.env.set_known_nodes(["node1", "node2", "node3"])
        self.before = dedent("""\
            totem {
                transport: udp

                interface {
                    broadcast: yes
                    mcastport: 1234
                    ttl: 128
                }
            }

            nodelist {
                node {
                    ring0_addr: node1-addr0
                    name: node1
                    nodeid: 1
                }

                node {
                    ring0_addr: node2-addr0
                    name: node2
                    nodeid: 2
                }

                node {
                    ring0_addr: node3-addr0
                    name: node3
                    nodeid: 3
                }
            }
            """
        )

    def test_success(self):
        after = dedent("""\
            totem {
                transport: udp

                interface {
                    ttl: 128
                    mcastaddr: 225.0.0.1
                }
            }

            nodelist {
                node {
                    ring0_addr: node1-addr0
                    name: node1
                    nodeid: 1
                }

                node {
                    ring0_addr: node2-addrA
                    name: node2
                    nodeid: 2
                }

                node {
                    ring0_addr: node3-addr0
                    name: node3
                    nodeid: 3
                }
            }
            """
        )
        patch_getaddrinfo(self, ["node2-addrA"])
        (self.config
            .corosync_conf.load_content(self.before)
            .env.push_corosync_conf(corosync_conf_text=after)
        )
        cluster.update_link(
            self.env_assist.get_env(),
            "0",
            {"node2": "node2-addrA"},
            {"mcastport": "", "broadcast": "0", "mcastaddr": "225.0.0.1"}
        )
        # Reports from pushing corosync.conf are produced in env. That code is
        # hidden in self.config.env.push_corosync_conf.
        self.env_assist.assert_reports([])

    def test_validation(self):
        patch_getaddrinfo(self, ["node3-addr0"])
        (self.config
            .corosync_conf.load_content(self.before)
        )
        self.env_assist.assert_raise_library_error(
            lambda: cluster.update_link(
                self.env_assist.get_env(),
                "0",
                {
                    "nodeX": "addr-new",
                    "node2": "",
                    "node1": "node3-addr0",
                },
                {
                    "wrong": "option",
                    "broadcast": "1",
                    "mcastaddr": "address",
                }
            ),
            []
        )
        self.env_assist.assert_reports(
            [
                fixture.error(
                    report_codes.INVALID_OPTION_VALUE,
                    option_value="address",
                    option_name="mcastaddr",
                    allowed_values="an IP address",
                    cannot_be_empty=False,
                    forbidden_characters=None,
                ),
                fixture.error(
                    report_codes.INVALID_OPTIONS,
                    option_names=["wrong"],
                    option_type="link",
                    allowed=[
                        "bindnetaddr", "broadcast", "mcastaddr", "mcastport",
                        "ttl",
                    ],
                    allowed_patterns=[],
                ),
                fixture.error(
                    report_codes.PREREQUISITE_OPTION_MUST_BE_DISABLED,
                    option_name="mcastaddr",
                    option_type="link",
                    prerequisite_name="broadcast",
                    prerequisite_type="link",
                ),
                fixture.error(
                    report_codes.NODE_NOT_FOUND,
                    node="nodeX",
                    searched_types=[],
                ),
                fixture.error(
                    report_codes.NODE_ADDRESSES_CANNOT_BE_EMPTY,
                    node_name_list=["node2"],
                ),
                fixture.error(
                    report_codes.NODE_ADDRESSES_UNRESOLVABLE,
                    force_code=report_codes.FORCE_NODE_ADDRESSES_UNRESOLVABLE,
                    address_list=["addr-new"],
                ),
                fixture.error(
                    report_codes.NODE_ADDRESSES_ALREADY_EXIST,
                    address_list=["node3-addr0"],
                ),
            ]
        )
