"""
Copyright 2020 The Magma Authors.

This source code is licensed under the BSD-style license found in the
LICENSE file in the root directory of this source tree.

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from datetime import datetime, timedelta, timezone
from unittest import TestCase, mock
from unittest.mock import Mock, patch

import lxml.etree as ET
from tests.test_utils.enb_acs_builder import (
    EnodebAcsStateMachineBuilder,
)
from tr069 import models
from tr069.rpc_methods import AutoConfigServer
from tr069.spyne_mods import Tr069Application, Tr069Soap11
from spyne import MethodContext
from spyne.server import ServerBase


class Tr069Test(TestCase):
    """ Tests for the TR-069 server """
    acs_to_cpe_queue = None
    cpe_to_acs_queue = None

    def setUp(self):
        # Set up the ACS
        self.enb_acs_manager = EnodebAcsStateMachineBuilder.build_acs_manager()
        self.handler = EnodebAcsStateMachineBuilder.build_acs_state_machine()
        AutoConfigServer.set_state_machine_manager(self.enb_acs_manager)

        def side_effect(*args, **_kwargs):
            msg = args[1]
            return msg

        self.p = patch.object(
            AutoConfigServer, '_handle_tr069_message',
            Mock(side_effect=side_effect),
        )
        self.p.start()

        self.app = Tr069Application(
            [AutoConfigServer],
            models.CWMP_NS,
            in_protocol=Tr069Soap11(validator='soft'),
            out_protocol=Tr069Soap11(),
        )

    def tearDown(self):
        self.p.stop()
        self.handler = None

    def _get_mconfig(self):
        return {
            "@type": "type.googleapis.com/magma.mconfig.EnodebD",
            "bandwidthMhz": 20,
            "specialSubframePattern": 7,
            "earfcndl": 44490,
            "logLevel": "INFO",
            "plmnidList": "00101",
            "pci": 260,
            "allowEnodebTransmit": False,
            "subframeAssignment": 2,
            "tac": 1,
        },

    def _get_service_config(self):
        return {
            "tr069": {
                "interface": "eth1",
                "port": 48080,
                "perf_mgmt_port": 8081,
                "public_ip": "192.88.99.142",
            },
            "reboot_enodeb_on_mme_disconnected": True,
            "s1_interface": "eth1",
        }

    def test_acs_manager_exception(self):
        """
        Test that an unexpected exception from the ACS SM manager will result
        in an empty response.
        """
        self.enb_acs_manager.handle_tr069_message = mock.MagicMock(
            side_effect=Exception('mock exception'),
        )
        # stop the patcher because we want to use the above MagicMock
        self.p.stop()
        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [b'']
        ctx, = server.generate_contexts(ctx)

        server.get_in_object(ctx)
        self.assertIsNone(ctx.in_error)

        server.get_out_object(ctx)
        self.assertIsNone(ctx.out_error)

        server.get_out_string(ctx)
        self.assertEqual(b''.join(ctx.out_string), b'')

        # start the patcher otherwise the p.stop() in tearDown will complain
        self.p.start()

    def test_parse_inform(self):
        """
        Test that example Inform RPC call can be parsed correctly
        """
        # Example TR-069 CPE->ACS RPC call. Copied from:
        # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
        cpe_string = b'''
            <soapenv:Envelope soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
                <soapenv:Header>
                    <cwmp:ID soapenv:mustUnderstand="1">0_THOM_TR69_ID</cwmp:ID>
                </soapenv:Header>
                <soapenv:Body>
                    <cwmp:Inform>
                        <DeviceId>
                            <Manufacturer>THOMSON</Manufacturer>
                            <OUI>00147F</OUI>
                            <ProductClass>SpeedTouch 780</ProductClass>
                            <SerialNumber>CP0611JTLNW</SerialNumber>
                        </DeviceId>
                        <Event soap:arrayType="cwmp:EventStruct[04]">
                            <EventStruct>
                                <EventCode>0 BOOTSTRAP</EventCode>
                                <CommandKey></CommandKey>
                            </EventStruct>
                            <EventStruct>
                                <EventCode>1 BOOT</EventCode>
                                <CommandKey></CommandKey>
                            </EventStruct>
                            <EventStruct>
                                <EventCode>2 PERIODIC</EventCode>
                                <CommandKey></CommandKey>
                            </EventStruct>
                            <EventStruct>
                                <EventCode>4 VALUE CHANGE</EventCode>
                                <CommandKey></CommandKey>
                            </EventStruct>
                        </Event>
                        <MaxEnvelopes>2</MaxEnvelopes>
                        <CurrentTime>1970-01-01T00:01:09Z</CurrentTime>
                        <RetryCount>05</RetryCount>
                        <ParameterList soap:arrayType="cwmp:ParameterValueStruct[12]">
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceSummary</Name>
                                <Value xsi:type="xsd:string">
                                    InternetGatewayDevice:1.1[] (Baseline:1, EthernetLAN:1, ADSLWAN:1, Bridging:1, Time:1, WiFiLAN:1)</Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceInfo.SpecVersion</Name>
                                <Value xsi:type="xsd:string">1.1</Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceInfo.HardwareVersion</Name>
                                <Value xsi:type="xsd:string">BANT-R</Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceInfo.SoftwareVersion</Name>
                                <Value xsi:type="xsd:string">6.2.35.0</Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceInfo.ProvisioningCode</Name>
                                <Value xsi:type="xsd:string"></Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceInfo.VendorConfigFile.1.Name</Name>
                                <Value xsi:type="xsd:string">MyCompanyName</Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceInfo.VendorConfigFile.1.Version</Name>
                                <Value xsi:type="xsd:string"></Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceInfo.VendorConfigFile.1.Date</Name>
                                <Value xsi:type="xsd:dateTime">0001-01-01T00:00:00</Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceInfo.VendorConfigFile.1.Description</Name>
                                <Value xsi:type="xsd:string">MyCompanyName</Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.ManagementServer.ConnectionRequestURL</Name>
                                <Value xsi:type="xsd:string">http://10.127.129.205:51005/</Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.ManagementServer.ParameterKey</Name>
                                <Value xsi:type="xsd:string"></Value>
                            </ParameterValueStruct>
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.WANDevice.1.WANConnectionDevice.1.WANIPConnection.1.ExternalIPAddress</Name>
                                <Value xsi:type="xsd:string">10.127.129.205</Value>
                            </ParameterValueStruct>
                        </ParameterList>
                    </cwmp:Inform>
                </soapenv:Body>
            </soapenv:Envelope>
            '''

        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [cpe_string]
        ctx, = server.generate_contexts(ctx)

        if ctx.in_error is not None:
            print('In error: %s' % ctx.in_error)
        self.assertEqual(ctx.in_error, None)

        server.get_in_object(ctx)

        self.assertEqual(ctx.in_object.DeviceId.OUI, '00147F')
        self.assertEqual(
            ctx.in_object.Event.EventStruct[0].EventCode, '0 BOOTSTRAP',
        )
        self.assertEqual(
            ctx.in_object.Event.EventStruct[2].EventCode, '2 PERIODIC',
        )
        self.assertEqual(ctx.in_object.MaxEnvelopes, 2)
        self.assertEqual(
            ctx.in_object.ParameterList.ParameterValueStruct[1].Name,
            'InternetGatewayDevice.DeviceInfo.SpecVersion',
        )
        self.assertEqual(
            str(ctx.in_object.ParameterList.ParameterValueStruct[1].Value), '1.1',
        )

    def test_parse_inform_cavium(self):
        """
        Test that example Inform RPC call can be parsed correctly from OC-LTE
        """
        cpe_string = b'''<?xml version="1.0" encoding="UTF-8"?>
        <SOAP-ENV:Envelope xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/envelope/" xmlns:SOAP-ENC="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
          <SOAP-ENV:Header>
            <cwmp:ID SOAP-ENV:mustUnderstand="1">CPE_1002</cwmp:ID>
          </SOAP-ENV:Header>
          <SOAP-ENV:Body SOAP-ENV:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">
            <cwmp:Inform>
              <DeviceId>
                <Manufacturer>Cavium, Inc.</Manufacturer>
                <OUI>000FB7</OUI>
                <ProductClass>Cavium eNB</ProductClass>
                <SerialNumber>10.18.104.79</SerialNumber>
              </DeviceId>
              <Event xsi:type="SOAP-ENC:Array" SOAP-ENC:arrayType="cwmp:EventStruct[1]">
                <EventStruct>
                  <EventCode>0 BOOTSTRAP</EventCode>
                  <CommandKey></CommandKey>
                </EventStruct>
              </Event>
              <MaxEnvelopes>1</MaxEnvelopes>
              <CurrentTime>1970-01-02T00:01:05.021239+00:00</CurrentTime>
              <RetryCount>2</RetryCount>
              <ParameterList xsi:type="SOAP-ENC:Array" SOAP-ENC:arrayType="cwmp:ParameterValueStruct[15]">
                <ParameterValueStruct>
                  <Name>Device.DeviceInfo.HardwareVersion</Name>
                  <Value xsi:type="xsd:string">1.0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.DeviceInfo.SoftwareVersion</Name>
                  <Value xsi:type="xsd:string">1.0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.DeviceInfo.AdditionalHardwareVersion</Name>
                  <Value xsi:type="xsd:string">1.0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.DeviceInfo.AdditionalSoftwareVersion</Name>
                  <Value xsi:type="xsd:string">1.0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.DeviceInfo.ProvisioningCode</Name>
                  <Value xsi:type="xsd:string">Cavium</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.ManagementServer.ParameterKey</Name>
                  <Value xsi:type="xsd:string"></Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.ManagementServer.ConnectionRequestURL</Name>
                  <Value xsi:type="xsd:string">http://192.88.99.253:8084/bucrhzjd</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.ManagementServer.UDPConnectionRequestAddress</Name>
                  <Value xsi:type="xsd:string"></Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.ManagementServer.NATDetected</Name>
                  <Value xsi:type="xsd:boolean">0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.IP.Diagnostics.UDPEchoConfig.PacketsReceived</Name>
                  <Value xsi:type="xsd:unsignedInt">0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.IP.Diagnostics.UDPEchoConfig.PacketsResponded</Name>
                  <Value xsi:type="xsd:unsignedInt">0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.IP.Diagnostics.UDPEchoConfig.BytesReceived</Name>
                  <Value xsi:type="xsd:unsignedInt">0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.IP.Diagnostics.UDPEchoConfig.BytesResponded</Name>
                  <Value xsi:type="xsd:unsignedInt">0</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.IP.Diagnostics.UDPEchoConfig.TimeFirstPacketReceived</Name>
                  <Value xsi:type="xsd:dateTime">1969-12-31T16:00:00.000000+00:00</Value>
                </ParameterValueStruct>
                <ParameterValueStruct>
                  <Name>Device.IP.Diagnostics.UDPEchoConfig.TimeLastPacketReceived</Name>
                  <Value xsi:type="xsd:dateTime">1969-12-31T16:00:00.000000+00:00</Value>
                </ParameterValueStruct>
              </ParameterList>
            </cwmp:Inform>
          </SOAP-ENV:Body>
        </SOAP-ENV:Envelope>
        '''

        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [cpe_string]
        ctx, = server.generate_contexts(ctx)

        if ctx.in_error is not None:
            print('In error: %s' % ctx.in_error)
        self.assertEqual(ctx.in_error, None)

        server.get_in_object(ctx)

        self.assertEqual(ctx.in_object.DeviceId.OUI, '000FB7')
        self.assertEqual(
            ctx.in_object.Event.EventStruct[0].EventCode, '0 BOOTSTRAP',
        )
        self.assertEqual(ctx.in_object.MaxEnvelopes, 1)
        self.assertEqual(
            ctx.in_object.ParameterList.ParameterValueStruct[1].Name,
            'Device.DeviceInfo.SoftwareVersion',
        )
        self.assertEqual(
            str(ctx.in_object.ParameterList.ParameterValueStruct[1].Value), '1.0',
        )

    def test_handle_transfer_complete(self):
        """
        Test that example TransferComplete RPC call can be parsed correctly, and
        response is correctly generated.
        """
        # Example TransferComplete CPE->ACS RPC request/response.
        # Manually created.
        cpe_string = b'''
            <soapenv:Envelope soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
                <soapenv:Header>
                    <cwmp:ID soapenv:mustUnderstand="1">1234</cwmp:ID>
                </soapenv:Header>
                <soapenv:Body>
                    <cwmp:TransferComplete>
                        <CommandKey>Downloading stuff</CommandKey>
                        <FaultStruct>
                            <FaultCode>0</FaultCode>
                            <FaultString></FaultString>
                        </FaultStruct>
                        <StartTime>2016-11-30T10:16:29Z</StartTime>
                        <CompleteTime>2016-11-30T10:17:05Z</CompleteTime>
                    </cwmp:TransferComplete>
                </soapenv:Body>
            </soapenv:Envelope>
            '''
        expected_acs_string = b'''
            <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
                <soapenv:Header>
                    <cwmp:ID soapenv:mustUnderstand="1">1234</cwmp:ID>
                </soapenv:Header>
                <soapenv:Body>
                    <cwmp:TransferCompleteResponse>
                    </cwmp:TransferCompleteResponse>
                </soapenv:Body>
            </soapenv:Envelope>
            '''

        self.p.stop()
        self.p.start()

        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [cpe_string]
        ctx, = server.generate_contexts(ctx)

        if ctx.in_error is not None:
            print('In error: %s' % ctx.in_error)
        self.assertEqual(ctx.in_error, None)

        server.get_in_object(ctx)
        self.assertEqual(ctx.in_error, None)

        server.get_out_object(ctx)
        self.assertEqual(ctx.out_error, None)

        output_msg = ctx.out_object[0]
        self.assertEqual(type(output_msg), models.TransferComplete)
        self.assertEqual(output_msg.CommandKey, 'Downloading stuff')
        self.assertEqual(output_msg.FaultStruct.FaultCode, 0)
        self.assertEqual(output_msg.FaultStruct.FaultString, '')
        self.assertEqual(
            output_msg.StartTime,
            datetime(
                2016, 11, 30, 10, 16, 29,
                tzinfo=timezone(timedelta(0)),
            ),
        )
        self.assertEqual(
            output_msg.CompleteTime,
            datetime(
                2016, 11, 30, 10, 17, 5,
                tzinfo=timezone(timedelta(0)),
            ),
        )

        server.get_out_string(ctx)
        self.assertEqual(ctx.out_error, None)

        xml_tree = XmlTree()
        match = xml_tree.xml_compare(
            xml_tree.convert_string_to_tree(b''.join(ctx.out_string)),
            xml_tree.convert_string_to_tree(expected_acs_string),
        )
        self.assertTrue(match)

    def test_parse_empty_http(self):
        """
        Test that empty HTTP message gets correctly mapped to 'EmptyHttp'
        function call
        """
        cpe_string = b''

        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [cpe_string]
        ctx, = server.generate_contexts(ctx)

        if ctx.in_error is not None:
            print('In error: %s' % ctx.in_error)

        self.assertEqual(ctx.in_error, None)
        self.assertEqual(ctx.function, AutoConfigServer.empty_http)

    def test_generate_empty_http(self):
        """
        Test that empty HTTP message is generated when setting output message
        name to 'EmptyHttp'
        """
        cpe_string = b''

        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [cpe_string]
        ctx, = server.generate_contexts(ctx)

        server.get_in_object(ctx)
        if ctx.in_error is not None:
            raise ctx.in_error

        server.get_out_object(ctx)
        if ctx.out_error is not None:
            raise ctx.out_error

        ctx.descriptor.out_message.Attributes.sub_name = 'EmptyHttp'
        ctx.out_object = [models.AcsToCpeRequests()]

        server.get_out_string(ctx)

        self.assertEqual(b''.join(ctx.out_string), b'')

    def test_generate_get_parameter_values_string(self):
        """
        Test that correct string is generated for SetParameterValues ACS->CPE
        request
        """
        # Example ACS->CPE RPC call. Copied from:
        # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
        # Following edits made:
        # - Change header ID value from 'null0' to 'null', to match magma
        #   default ID
        expected_acs_string = b'''
        <soapenv:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:cwmp="urn:dslforum-org:cwmp-1-0" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
            <soapenv:Header>
                <cwmp:ID soapenv:mustUnderstand="1">null</cwmp:ID>
            </soapenv:Header>
            <soapenv:Body>
                <cwmp:GetParameterValues>
                    <ParameterNames soap:arrayType="xsd:string[1]">
                        <string>foo</string>
                    </ParameterNames>
                </cwmp:GetParameterValues>
            </soapenv:Body>
        </soapenv:Envelope>
        '''

        names = ['foo']
        request = models.GetParameterValues()
        request.ParameterNames = models.ParameterNames()
        request.ParameterNames.arrayType = 'xsd:string[%d]' \
            % len(names)
        request.ParameterNames.string = []
        for name in names:
            request.ParameterNames.string.append(name)

        request.ParameterKey = 'null'

        def side_effect(*args, **_kwargs):
            ctx = args[0]
            ctx.out_header = models.ID(mustUnderstand='1')
            ctx.out_header.Data = 'null'
            ctx.descriptor.out_message.Attributes.sub_name = \
                request.__class__.__name__
            return AutoConfigServer._generate_acs_to_cpe_request_copy(request)

        self.p.stop()
        self.p = patch.object(
            AutoConfigServer, '_handle_tr069_message',
            side_effect=side_effect,
        )
        self.p.start()

        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [b'']
        ctx, = server.generate_contexts(ctx)

        server.get_in_object(ctx)
        if ctx.in_error is not None:
            raise ctx.in_error

        server.get_out_object(ctx)
        if ctx.out_error is not None:
            raise ctx.out_error

        server.get_out_string(ctx)

        xml_tree = XmlTree()
        match = xml_tree.xml_compare(
            xml_tree.convert_string_to_tree(b''.join(ctx.out_string)),
            xml_tree.convert_string_to_tree(expected_acs_string),
        )
        self.assertTrue(match)

    def test_generate_set_parameter_values_string(self):
        """
        Test that correct string is generated for SetParameterValues ACS->CPE
        request
        """
        # Example ACS->CPE RPC call. Copied from:
        # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
        # Following edits made:
        # - Change header ID value from 'null0' to 'null', to match magma
        #   default ID
        expected_acs_string = b'''
        <soapenv:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:cwmp="urn:dslforum-org:cwmp-1-0" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
            <soapenv:Header>
                <cwmp:ID soapenv:mustUnderstand="1">null</cwmp:ID>
            </soapenv:Header>
            <soapenv:Body>
                <cwmp:SetParameterValues>
                    <ParameterList soap:arrayType="cwmp:ParameterValueStruct[4]">
                        <ParameterValueStruct>
                            <Name>InternetGatewayDevice.ManagementServer.PeriodicInformEnable</Name>
                            <Value xsi:type="xsd:boolean">1</Value>
                        </ParameterValueStruct>
                        <ParameterValueStruct>
                            <Name>InternetGatewayDevice.ManagementServer.ConnectionRequestUsername</Name>
                            <Value xsi:type="xsd:string">00147F-SpeedTouch780-CP0611JTLNW</Value>
                        </ParameterValueStruct>
                        <ParameterValueStruct>
                            <Name>InternetGatewayDevice.ManagementServer.ConnectionRequestPassword</Name>
                            <Value xsi:type="xsd:string">98ff55fb377bf724c625f60dec448646</Value>
                        </ParameterValueStruct>
                        <ParameterValueStruct>
                            <Name>InternetGatewayDevice.ManagementServer.PeriodicInformInterval</Name>
                            <Value xsi:type="xsd:unsignedInt">60</Value>
                        </ParameterValueStruct>
                    </ParameterList>
                    <ParameterKey xsi:type="xsd:string">SetParameter1</ParameterKey>
                </cwmp:SetParameterValues>
            </soapenv:Body>
        </soapenv:Envelope>
        '''

        request = models.SetParameterValues()

        request.ParameterList = \
            models.ParameterValueList(arrayType='cwmp:ParameterValueStruct[4]')
        request.ParameterList.ParameterValueStruct = []

        param = models.ParameterValueStruct()
        param.Name = 'InternetGatewayDevice.ManagementServer.PeriodicInformEnable'
        param.Value = models.anySimpleType(type='xsd:boolean')
        param.Value.Data = '1'
        request.ParameterList.ParameterValueStruct.append(param)

        param = models.ParameterValueStruct()
        param.Name = 'InternetGatewayDevice.ManagementServer.ConnectionRequestUsername'
        param.Value = models.anySimpleType(type='xsd:string')
        param.Value.Data = '00147F-SpeedTouch780-CP0611JTLNW'
        request.ParameterList.ParameterValueStruct.append(param)

        param = models.ParameterValueStruct()
        param.Name = 'InternetGatewayDevice.ManagementServer.ConnectionRequestPassword'
        param.Value = models.anySimpleType(type='xsd:string')
        param.Value.Data = '98ff55fb377bf724c625f60dec448646'
        request.ParameterList.ParameterValueStruct.append(param)

        param = models.ParameterValueStruct()
        param.Name = 'InternetGatewayDevice.ManagementServer.PeriodicInformInterval'
        param.Value = models.anySimpleType(type='xsd:unsignedInt')
        param.Value.Data = '60'
        request.ParameterList.ParameterValueStruct.append(param)

        request.ParameterKey = models.ParameterKeyType()
        request.ParameterKey.type = 'xsd:string'
        request.ParameterKey.Data = 'SetParameter1'

        def side_effect(*args, **_kwargs):
            ctx = args[0]
            ctx.out_header = models.ID(mustUnderstand='1')
            ctx.out_header.Data = 'null'
            ctx.descriptor.out_message.Attributes.sub_name = request.__class__.__name__
            return request

        self.p.stop()
        self.p = patch.object(
            AutoConfigServer, '_handle_tr069_message',
            Mock(side_effect=side_effect),
        )
        self.p.start()

        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [b'']
        ctx, = server.generate_contexts(ctx)

        server.get_in_object(ctx)
        if ctx.in_error is not None:
            raise ctx.in_error

        server.get_out_object(ctx)
        if ctx.out_error is not None:
            raise ctx.out_error

        server.get_out_string(ctx)

        xml_tree = XmlTree()
        NS_SOAP11_ENC = 'soap11enc'
        NS_SOAP11_ENV = 'soap11env'
        xml_str = b''.join(ctx.out_string)
        # Get the namespaces and validate the soap enc and env prefix are right
        nsmap = xml_tree.get_ns(xml_str)
        self.assertTrue(NS_SOAP11_ENC in nsmap.keys())
        self.assertTrue(NS_SOAP11_ENV in nsmap.keys())

        match = xml_tree.xml_compare(
            xml_tree.convert_string_to_tree(xml_str),
            xml_tree.convert_string_to_tree(expected_acs_string),
        )
        self.assertTrue(match)

    def test_parse_fault_response(self):
        """ Tests that a fault response from CPE is correctly parsed. """
        # Example CPE->ACS fault response. Copied from:
        # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
        cpe_string = b'''
        <soapenv:Envelope soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
         <soapenv:Header>
        <cwmp:ID soapenv:mustUnderstand="1">1031422463</cwmp:ID>
         </soapenv:Header>
         <soapenv:Body>
           <soapenv:Fault>
            <faultcode>Client</faultcode>
            <faultstring>CWMP fault</faultstring>
            <detail>
             <cwmp:Fault>
              <FaultCode>9003</FaultCode>
              <FaultString>Invalid arguments</FaultString>
              <SetParameterValuesFault>
               <ParameterName>InternetGatewayDevice.WANDevice.1.WANConnectionDevice.3.WANPPPConnection.1.Password</ParameterName>
               <FaultCode>9003</FaultCode>
               <FaultString>Invalid arguments</FaultString>
              </SetParameterValuesFault>
              <SetParameterValuesFault>
               <ParameterName>InternetGatewayDevice.WANDevice.1.WANConnectionDevice.3.WANPPPConnection.1.Username</ParameterName>
               <FaultCode>9003</FaultCode>
               <FaultString>Invalid arguments</FaultString>
              </SetParameterValuesFault>
             </cwmp:Fault>
            </detail>
           </soapenv:Fault>
         </soapenv:Body>
        </soapenv:Envelope>
        '''
        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [cpe_string]
        ctx, = server.generate_contexts(ctx)
        server.get_in_object(ctx)
        if ctx.in_error is not None:
            raise ctx.in_error

        # Calls function to receive and process message
        server.get_out_object(ctx)

        output_msg = ctx.out_object[0]
        self.assertEqual(type(output_msg), models.Fault)
        self.assertEqual(output_msg.FaultCode, 9003)
        self.assertEqual(output_msg.FaultString, 'Invalid arguments')
        self.assertEqual(
            output_msg.SetParameterValuesFault[1].ParameterName,
            'InternetGatewayDevice.WANDevice.1.WANConnectionDevice.3.WANPPPConnection.1.Username',
        )
        self.assertEqual(output_msg.SetParameterValuesFault[1].FaultCode, 9003)
        self.assertEqual(
            output_msg.SetParameterValuesFault[1].FaultString,
            'Invalid arguments',
        )

    def test_parse_hex_values(self):
        """
        Test that non-utf-8 hex values can be parsed without error
        """
        # Example TR-069 CPE->ACS RPC call. Copied from:
        # http://djuro82.blogspot.com/2011/05/tr-069-cpe-provisioning.html
        cpe_string = b'''
            <soapenv:Envelope soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:soap="http://schemas.xmlsoap.org/soap/encoding/" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cwmp="urn:dslforum-org:cwmp-1-0">
                <soapenv:Header>
                    <cwmp:ID soapenv:mustUnderstand="1">0_THOM_TR69_ID</cwmp:ID>
                </soapenv:Header>
                <soapenv:Body>
                    <cwmp:Inform>
                        <DeviceId>
                            <Manufacturer>THOMSON</Manufacturer>
                            <OUI>00147F</OUI>
                            <ProductClass>SpeedTouch 780</ProductClass>
                            <SerialNumber>CP0611JTLNW</SerialNumber>
                        </DeviceId>
                        <Event soap:arrayType="cwmp:EventStruct[04]">
                            <EventStruct>
                                <EventCode>0 BOOTSTRAP</EventCode>
                                <CommandKey></CommandKey>
                            </EventStruct>
                            <EventStruct>
                                <EventCode>1 BOOT</EventCode>
                                <CommandKey></CommandKey>
                            </EventStruct>
                            <EventStruct>
                                <EventCode>2 PERIODIC</EventCode>
                                <CommandKey></CommandKey>
                            </EventStruct>
                            <EventStruct>
                                <EventCode>4 VALUE CHANGE</EventCode>
                                <CommandKey></CommandKey>
                            </EventStruct>
                        </Event>
                        <MaxEnvelopes>2</MaxEnvelopes>
                        <CurrentTime>1970-01-01T00:01:09Z</CurrentTime>
                        <RetryCount>05</RetryCount>
                        <ParameterList soap:arrayType="cwmp:ParameterValueStruct[12]">
                            <ParameterValueStruct>
                                <Name>InternetGatewayDevice.DeviceSummary</Name>
                                <Value xsi:type="xsd:string">
                                    \xff\xff\xff\xff\xff</Value>
                            </ParameterValueStruct>
                        </ParameterList>
                    </cwmp:Inform>
                </soapenv:Body>
            </soapenv:Envelope>
            '''

        server = ServerBase(self.app)

        ctx = MethodContext(server, MethodContext.SERVER)
        ctx.in_string = [cpe_string]
        ctx, = server.generate_contexts(ctx)

        if ctx.in_error is not None:
            print('In error: %s' % ctx.in_error)
        self.assertEqual(ctx.in_error, None)

        server.get_in_object(ctx)


class XmlTree():

    @staticmethod
    def convert_string_to_tree(xmlString):

        return ET.fromstring(xmlString)

    @staticmethod
    def get_ns(xmlString):
        return ET.fromstring(xmlString).nsmap

    def xml_compare(self, x1, x2, excludes=None):
        """
        Compares two xml etrees
        :param x1: the first tree
        :param x2: the second tree
        :param excludes: list of string of attributes to exclude from comparison
        :return:
            True if both files match
        """
        excludes = [] if excludes is None else excludes

        if x1.tag != x2.tag:
            print('Tags do not match: %s and %s' % (x1.tag, x2.tag))
            return False
        for name, value in x1.attrib.items():
            if name not in excludes:
                if x2.attrib.get(name) != value:
                    print(
                        'Attributes do not match: %s=%r, %s=%r'
                        % (name, value, name, x2.attrib.get(name)),
                    )
                    return False
        for name in x2.attrib.keys():
            if name not in excludes:
                if name not in x1.attrib:
                    print(
                        'x2 has an attribute x1 is missing: %s'
                        % name,
                    )
                    return False
        if not self.text_compare(x1.text, x2.text):
            print('text: %r != %r' % (x1.text, x2.text))
            return False
        if not self.text_compare(x1.tail, x2.tail):
            print('tail: %r != %r' % (x1.tail, x2.tail))
            return False
        cl1 = x1.getchildren()
        cl2 = x2.getchildren()
        if len(cl1) != len(cl2):
            print(
                'children length differs, %i != %i'
                % (len(cl1), len(cl2)),
            )
            return False
        i = 0
        for c1, c2 in zip(cl1, cl2):
            i += 1
            if c1.tag not in excludes:
                if not self.xml_compare(c1, c2, excludes):
                    print(
                        'children %i do not match: %s'
                        % (i, c1.tag),
                    )
                    return False
        return True

    def text_compare(self, t1, t2):
        """
        Compare two text strings
        :param t1: text one
        :param t2: text two
        :return:
            True if a match
        """
        if not t1 and not t2:
            return True
        if t1 == '*' or t2 == '*':
            return True
        return (t1 or '').strip() == (t2 or '').strip()