From f7408882e31f8e0e18605a3070c8b4ed57699e16 Mon Sep 17 00:00:00 2001 From: Piotr Maslanka <piotr.maslanka@henrietta.com.pl> Date: Tue, 10 Jan 2017 03:50:57 +0100 Subject: [PATCH] tests --- tests/test_connection/__init__.py | 2 - tests/test_connection/test_state.py | 36 ------------ .../test_definitions/test_frames.py | 7 +-- tests/test_uplink/__init__.py | 2 - tests/test_uplink/test_basic.py | 57 ------------------- 5 files changed, 3 insertions(+), 101 deletions(-) delete mode 100644 tests/test_connection/__init__.py delete mode 100644 tests/test_connection/test_state.py delete mode 100644 tests/test_uplink/__init__.py delete mode 100644 tests/test_uplink/test_basic.py diff --git a/tests/test_connection/__init__.py b/tests/test_connection/__init__.py deleted file mode 100644 index 9f2b35b..0000000 --- a/tests/test_connection/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -# coding=UTF-8 -from __future__ import absolute_import, division, print_function diff --git a/tests/test_connection/test_state.py b/tests/test_connection/test_state.py deleted file mode 100644 index 2d36b98..0000000 --- a/tests/test_connection/test_state.py +++ /dev/null @@ -1,36 +0,0 @@ -# coding=UTF-8 -from __future__ import absolute_import, division, print_function -import unittest - - -from coolamqp.connection.state import Broker -from coolamqp.objects import NodeDefinition -from coolamqp.uplink import ListenerThread, Connection, Handshaker -import socket -import time - - -def newc(): - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect(('127.0.0.1', 5672)) - s.settimeout(0) - s.send('AMQP\x00\x00\x09\x01') - return s - - -NODE = NodeDefinition('127.0.0.1', 5672, 'user', 'user') - - -class TestState(unittest.TestCase): - def test_basic(self): - lt = ListenerThread() - lt.start() - - broker = Broker(Connection(newc(), lt), NODE) - - ord = broker.connect() - ord.wait() - - time.sleep(100) - - lt.terminate() diff --git a/tests/test_framing/test_definitions/test_frames.py b/tests/test_framing/test_definitions/test_frames.py index c805647..8f57d0a 100644 --- a/tests/test_framing/test_definitions/test_frames.py +++ b/tests/test_framing/test_definitions/test_frames.py @@ -33,14 +33,13 @@ class TestShitSerializesRight(unittest.TestCase): hdr = AMQPHeaderFrame(0, 60, 0, 0, a_cpl) hdr.write_to(buf) - s = b'\x00\x00\x00\x00' + \ + s = b'\x00\x3C\x00\x00' + \ b'\x00\x00\x00\x00\x00\x00\x00\x00' + \ b'\x80\x00\x0Atext/plain' + s = chr(FRAME_HEADER) + b'\x00\x00' + \ struct.pack('!L', len(s)) + s + chr(FRAME_END) - self.assertEquals(buf.getvalue(), - - ) + self.assertEquals(buf.getvalue(), s) diff --git a/tests/test_uplink/__init__.py b/tests/test_uplink/__init__.py deleted file mode 100644 index 9f2b35b..0000000 --- a/tests/test_uplink/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -# coding=UTF-8 -from __future__ import absolute_import, division, print_function diff --git a/tests/test_uplink/test_basic.py b/tests/test_uplink/test_basic.py deleted file mode 100644 index 034d7e9..0000000 --- a/tests/test_uplink/test_basic.py +++ /dev/null @@ -1,57 +0,0 @@ -# coding=UTF-8 -from __future__ import absolute_import, division, print_function -import unittest - -from coolamqp.uplink import ListenerThread, Connection, Handshaker -import socket -import time - - -def newc(): - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect(('127.0.0.1', 5672)) - s.settimeout(0) - s.send('AMQP\x00\x00\x09\x01') - return s - - -class TestBasic(unittest.TestCase): - def test_gets_connectionstart(self): - - hnd_ok = {'ok': False} - def hnd_suc(): - hnd_ok['ok'] = True - - lt = ListenerThread() - lt.start() - - con = Connection(newc(), lt) - - Handshaker(con, 'user', 'user', '/', hnd_suc, lambda: None) - con.start() - - time.sleep(5) - - lt.terminate() - self.assertTrue(hnd_ok['ok']) - - - def test_heartbeats(self): - - hnd_ok = {'ok': False} - def hnd_suc(): - hnd_ok['ok'] = True - - lt = ListenerThread() - lt.start() - - con = Connection(newc(), lt) - - Handshaker(con, 'user', 'user', '/', hnd_suc, lambda: None, 3) - con.start() - - time.sleep(20) - - self.assertFalse(con.failed) - lt.terminate() - self.assertTrue(hnd_ok['ok']) -- GitLab