added executable script to run EPP Gate server from command line

Author: vesellov

bin/epp-gate

@@ -0,0 +1,8 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import re
+import sys
+from epp.rpc_server import main
+if __name__ == '__main__':
+    sys.argv[0] = re.sub(r'(-script\.pyw|\.exe)?$', '', sys.argv[0])
+    sys.exit(int(not main()))

requirements.txt

@@ -10,3 +10,4 @@ pyparsing==2.4.7
 pytest==6.2.3
 soupsieve==2.2.1
 toml==0.10.2
+pika==1.2.0

setup.py

@@ -7,6 +7,7 @@
     author_email='[email protected]',
     packages=find_packages(where='src'),
     package_dir={'': 'src'},
+    scripts=['bin/epp-gate', ],
     description='Python-based Extensible Provisioning Protocol (EPP) client',
     long_description=(
         "The library provides an interface to the Extensible Provisioning "

tests/epp/test_client.py

@@ -35,6 +35,13 @@
 <trID><svTRID>1618435120843</svTRID></trID></response></epp>'''
 
 
+sample_unknown_command_response = b'''<?xml version="1.0" encoding="UTF-8"?>
+<epp xmlns="urn:ietf:params:xml:ns:epp-1.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+xsi:schemaLocation="urn:ietf:params:xml:ns:epp-1.0 epp-1.0.xsd"><response><result code="2000">
+<msg>Unknown command</msg></result>
+<trID><svTRID>1618663648910</svTRID></trID></response></epp>'''
+
+
 def conn():
     logging.getLogger('epp.client').setLevel(logging.DEBUG)
     return client.EPPConnection(
@@ -47,6 +54,10 @@ def conn():
     )
 
 
+def fake_response(xml_response):
+    return client.int_to_net(len(xml_response) + 4, client.get_format_32()) + xml_response
+
+
 class TestEPPConnection(object):
 
     def test_init(self):
@@ -94,8 +105,8 @@ def test_greeting_failed(self, mock_wrap_socket, mock_socket_connect):
     def test_login_failed(self, mock_wrap_socket, mock_socket_connect):
         mock_socket_connect.return_value = True
         fake_stream = io.BytesIO(
-            client.int_to_net(len(sample_greeting) + 4, client.get_format_32()) + sample_greeting +
-            client.int_to_net(18 + 4, client.get_format_32()) + b'bad login response')
+            fake_response(sample_greeting) +
+            fake_response(b'bad login response'))
         setattr(fake_stream, 'send', lambda _: True)
         mock_wrap_socket.return_value = fake_stream
         with pytest.raises(AttributeError):
@@ -106,8 +117,8 @@ def test_login_failed(self, mock_wrap_socket, mock_socket_connect):
     def test_open_success(self, mock_wrap_socket, mock_socket_connect):
         mock_socket_connect.return_value = True
         fake_stream = io.BytesIO(
-            client.int_to_net(len(sample_greeting) + 4, client.get_format_32()) + sample_greeting +
-            client.int_to_net(len(sample_login_response) + 4, client.get_format_32()) + sample_login_response)
+            fake_response(sample_greeting) +
+            fake_response(sample_login_response))
         setattr(fake_stream, 'send', lambda _: True)
         mock_wrap_socket.return_value = fake_stream
         assert conn().open() is True
@@ -117,9 +128,9 @@ def test_open_success(self, mock_wrap_socket, mock_socket_connect):
     def test_close_failed(self, mock_wrap_socket, mock_socket_connect):
         mock_socket_connect.return_value = True
         fake_stream = io.BytesIO(
-            client.int_to_net(len(sample_greeting) + 4, client.get_format_32()) + sample_greeting +
-            client.int_to_net(len(sample_login_response) + 4, client.get_format_32()) + sample_login_response +
-            client.int_to_net(19 + 4, client.get_format_32()) + b'bad logout response')
+            fake_response(sample_greeting) +
+            fake_response(sample_login_response) +
+            fake_response(b'bad logout response'))
         setattr(fake_stream, 'send', lambda _: True)
         mock_wrap_socket.return_value = fake_stream
         c = conn()
@@ -132,9 +143,9 @@ def test_close_failed(self, mock_wrap_socket, mock_socket_connect):
     def test_close_success(self, mock_wrap_socket, mock_socket_connect):
         mock_socket_connect.return_value = True
         fake_stream = io.BytesIO(
-            client.int_to_net(len(sample_greeting) + 4, client.get_format_32()) + sample_greeting +
-            client.int_to_net(len(sample_login_response) + 4, client.get_format_32()) + sample_login_response +
-            client.int_to_net(len(sample_logout_response) + 4, client.get_format_32()) + sample_logout_response)
+            fake_response(sample_greeting) +
+            fake_response(sample_login_response) +
+            fake_response(sample_logout_response))
         setattr(fake_stream, 'send', lambda _: True)
         mock_wrap_socket.return_value = fake_stream
         c = conn()
@@ -146,12 +157,34 @@ def test_close_success(self, mock_wrap_socket, mock_socket_connect):
     def test_call(self, mock_wrap_socket, mock_socket_connect):
         mock_socket_connect.return_value = True
         fake_stream = io.BytesIO(
-            client.int_to_net(len(sample_greeting) + 4, client.get_format_32()) + sample_greeting +
-            client.int_to_net(len(sample_login_response) + 4, client.get_format_32()) + sample_login_response +
-            client.int_to_net(len(sample_logout_response) + 4, client.get_format_32()) + sample_logout_response)
+            fake_response(sample_greeting) +
+            fake_response(sample_login_response) +
+            fake_response(sample_unknown_command_response) +
+            fake_response(sample_logout_response))
         setattr(fake_stream, 'send', lambda _: True)
         mock_wrap_socket.return_value = fake_stream
         c = conn()
         assert c.open() is True
-        c.call(cmd='<?xml version="1.0" encoding="UTF-8"?><test></test>')
+        assert c.call(cmd='<?xml version="1.0" encoding="UTF-8"?><test></test>') == sample_unknown_command_response
+        assert c.close() is True
 
+    @mock.patch('socket.socket.connect')
+    @mock.patch('ssl.wrap_socket')
+    def test_call_with_soup(self, mock_wrap_socket, mock_socket_connect):
+        mock_socket_connect.return_value = True
+        fake_stream = io.BytesIO(
+            fake_response(sample_greeting) +
+            fake_response(sample_login_response) +
+            fake_response(sample_unknown_command_response) +
+            fake_response(sample_logout_response))
+        setattr(fake_stream, 'send', lambda _: True)
+        mock_wrap_socket.return_value = fake_stream
+        c = conn()
+        assert c.open() is True
+        resp = c.call(
+            cmd='<?xml version="1.0" encoding="UTF-8"?><test></test>',
+            soup=True,
+        )
+        assert resp.find('result').get('code') == '2000'
+        assert resp.find('msg').text == 'Unknown command'
+        assert c.close() is True