"""Unittests for Controller class in controller.py"""
import sys
import os
sys.path.insert(0, os.path.abspath(os.path.join(__file__, "../../../")))
from gstswitch.controller import Controller
from gstswitch.exception import ConnectionReturnError
import pytest
from mock import Mock
from gstswitch.connection import Connection
from gi.repository import GLib
[docs]class TestAddress(object):
"""Test the address parameter"""
[docs] def test_address_null(self):
"""Test if address is null"""
address = ['', None, [], {}]
for addr in address:
with pytest.raises(ValueError):
Controller(address=addr)
[docs] def test_address_colon(self):
"""Test if address has no colon"""
address = 'abcdefghijk'
with pytest.raises(ValueError):
Controller(address=address)
[docs] def test_address_normal(self):
"""Test if address is valid"""
address = ['unix:abstract=gstswitch', 'unix:temp=/tmp/abcd/xyz']
for addr in address:
conn = Controller(address=addr)
assert conn.address == addr
[docs]class TestBusName(object):
"""Test bus_name parameter"""
[docs] def test_normal(self):
"""Test when bus_name is not null"""
names = ['', 'abcd', 12345]
for bus in names:
conn = Controller(bus_name=bus)
assert conn.bus_name == str(bus)
[docs] def test_normal_none(self):
"""Test when bus_name is null"""
name = None
conn = Controller(bus_name=name)
assert conn.bus_name == name
[docs]class TestObjectPath(object):
"""Test object_path parameter"""
[docs] def test_object_path_blank(self):
"""Test when the object_path is null"""
paths = [None, '', {}, []]
for object_path in paths:
with pytest.raises(ValueError):
Controller(object_path=object_path)
[docs] def test_object_path_slash(self):
"""Test when object_path doesn't have slash in start"""
object_path = 'a/////////'
with pytest.raises(ValueError):
Controller(object_path=object_path)
[docs] def test_object_path_normal(self):
"""Test when object_path is valid"""
object_path = "/info/duzy/gst/switch/SwitchController"
conn = Controller(object_path=object_path)
assert conn.object_path == object_path
[docs]class TestInterface(object):
"""Test the default_interface parameter"""
[docs] def test_interface_none(self):
"""Test when the default_interface is null"""
default_interface = [None, '', [], {}]
for interface in default_interface:
with pytest.raises(ValueError):
Controller(default_interface=interface)
[docs] def test_interface_dot(self):
"""Test when the default_interface has <2 dots"""
default_interface = ['.', 'info.', 'info']
for interface in default_interface:
with pytest.raises(ValueError):
Controller(default_interface=interface)
[docs] def test_interface_normal(self):
"""Test when the interface is valid"""
default_interface = "info.duzy.gst.switch.SwitchControllerInterface"
conn = Controller(default_interface=default_interface)
assert default_interface == conn.default_interface
[docs]class TestEstablishConnection(object):
"""Test the establish_connection method"""
[docs] def test_normal(self, monkeypatch):
"""Test if the parameters are valid"""
monkeypatch.setattr(Connection, 'connect_dbus', Mock())
controller = Controller(address='unix:abstract=abcd')
controller.establish_connection()
assert controller.connection is not None
[docs]class MockConnection(object):
"""A class which mocks the Connection class"""
def __init__(self, mode):
self.mode = mode
[docs] def get_compose_port(self):
"""mock of get_compose_port"""
if self.mode is False:
return GLib.Variant('(i)', (3001,))
else:
return (0,)
[docs] def get_encode_port(self):
"""mock of get_encode_port"""
if self.mode is False:
return GLib.Variant('(i)', (3002,))
else:
return (0,)
[docs] def get_audio_port(self):
"""mock of get_audio_port"""
if self.mode is False:
return GLib.Variant('(i)', (4000,))
else:
return (0,)
[docs] def get_preview_ports(self):
"""mock of get_preview_ports"""
if self.mode is False:
return GLib.Variant('(s)', ('[(3002, 1, 7), (3003, 1, 8)]',))
else:
return (0,)
[docs] def set_composite_mode(self, mode):
"""mock of set_composite_mode"""
if self.mode is False:
return GLib.Variant('(b)', (True,))
else:
return (False,)
[docs] def set_encode_mode(self, mode):
"""mock of get_set_encode_mode"""
if self.mode is False:
return GLib.Variant('(b)', (True,))
else:
return (True,)
[docs] def new_record(self):
"""mock of new_record"""
if self.mode is False:
return GLib.Variant('(b)', (True,))
else:
return (True,)
[docs] def adjust_pip(self, xpos, ypos, width, height):
"""mock of adjust_pip"""
if self.mode is False:
return GLib.Variant('(u)', (1,))
else:
return (1,)
[docs] def switch(self, channel, port):
"""mock of switch"""
if self.mode is False:
return GLib.Variant('(b)', (True,))
else:
return (True,)
[docs] def click_video(self, xpos, ypos, width, height):
"""mock of click_video"""
if self.mode is False:
return GLib.Variant('(b)', (True,))
else:
return (True,)
[docs] def mark_face(self, face):
"""mock of mark_face"""
pass
[docs] def mark_tracking(self, face):
"""mock of mark_tracking"""
pass
[docs]class TestGetComposePort(object):
"""Test the get_compose_port method"""
[docs] def test_unpack(self):
"""Test when values cant unpack"""
controller = Controller(address='unix:abstract=abcdefghijk')
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.get_compose_port()
[docs] def test_normal_unpack(self):
"""Test when valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.connection = MockConnection(False)
assert controller.get_compose_port() == 3001
[docs]class TestGetEncodePort(object):
"""Test the get_encode_port method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcdefghijk')
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.get_encode_port()
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.connection = MockConnection(False)
assert controller.get_encode_port() == 3002
[docs]class TestGetAudioPort(object):
""" Test the get_audio_port method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcdefghijk')
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.get_audio_port()
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.connection = MockConnection(False)
assert controller.get_audio_port() == 4000
[docs]class TestGetPreviewPorts(object):
"""Test the get_preview_ports method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcdefghijk')
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.get_preview_ports()
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.connection = MockConnection(False)
controller.parse_preview_ports = Mock(return_value=[3001, 3002])
assert controller.get_preview_ports() == [3001, 3002]
[docs]class TestSetCompositeMode(object):
"""Test the set_composite_mode method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcdefghijk')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.set_composite_mode(1)
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(False)
assert controller.set_composite_mode(1) is True
[docs]class TestSetEncodeMode(object):
"""Test the set_encode_mode method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcde')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.set_encode_mode(1)
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(False)
assert controller.set_encode_mode(1) is True
[docs]class TestNewRecord(object):
"""Test the new_record method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcde')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.new_record()
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(False)
assert controller.new_record() is True
[docs]class TestAdjustPIP(object):
"""Test the adjust_pip method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcde')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.adjust_pip(1, 2, 3, 4)
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(False)
assert controller.adjust_pip(1, 2, 3, 4) == 1
[docs]class TestSwitch(object):
"""Test the switch method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcde')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.switch(1, 2)
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(False)
assert controller.switch(1, 2) is True
[docs]class TestClickVideo(object):
"""Test the click_video method"""
[docs] def test_unpack(self):
"""Test if unpack fails"""
controller = Controller(address='unix:abstract=abcde')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(True)
with pytest.raises(ConnectionReturnError):
controller.click_video(1, 2, 3, 4)
[docs] def test_normal_unpack(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcdef')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(False)
assert controller.click_video(1, 2, 3, 4) is True
[docs]class TestMarkFaces(object):
"""Tes the mark_face method"""
[docs] def test_normal(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcde')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(True)
face = [(1, 2, 3, 4), (1, 1, 1, 1)]
controller.mark_face(face)
[docs]class TestMarkTracking(object):
"""Test the mark_tracking method"""
[docs] def test_normal(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcde')
controller.establish_connection = Mock(return_value=None)
controller.connection = MockConnection(True)
face = [(1, 2, 3, 4), (1, 1, 1, 1)]
controller.mark_tracking(face)
[docs]class TestParsePreviewPorts(object):
"""Test the parse_preview_ports class method"""
[docs] def test_value_error(self):
"""Test if invalid"""
controller = Controller(address='unix:abstract=abcde')
test = 1234
with pytest.raises(ConnectionReturnError):
controller.parse_preview_ports(test)
[docs] def test_syntax_error(self):
"""Test if syntax error detected"""
controller = Controller(address='unix:abstract=abcde')
test = '{}[]'
with pytest.raises(ConnectionReturnError):
controller.parse_preview_ports(test)
[docs] def test_normal(self):
"""Test if valid"""
controller = Controller(address='unix:abstract=abcde')
test = '[(1, 2, 3), (2, 2, 2)]'
assert controller.parse_preview_ports(test) == [1, 2]