diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index d053e22..364bece 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -30,4 +30,4 @@ jobs: python -m pip install --upgrade cryptography requests colorama - name: "Run test.py and tests.py on ${{ matrix.python-version }}" - run: "python -m test.py && python -m tests" \ No newline at end of file + run: "python -m test.py && python -m tests" diff --git a/tests.py b/tests.py index b6e3c5e..f485178 100755 --- a/tests.py +++ b/tests.py @@ -7,9 +7,7 @@ from unittest.mock import MagicMock # Python 3 except ImportError: from mock import MagicMock # py2 use https://pypi.python.org/pypi/mock -from hashlib import md5 import json -import logging import struct # Enable info logging to see version information @@ -24,173 +22,122 @@ mock_byte_encoding = 'utf-8' -def compare_json_strings(json1, json2, ignoring_keys=None): - json1 = json.loads(json1) - json2 = json.loads(json2) - - if ignoring_keys is not None: - for key in ignoring_keys: - json1[key] = json2[key] - - return json.dumps(json1, sort_keys=True) == json.dumps(json2, sort_keys=True) - -def check_data_frame(data, expected_prefix, encrypted=True): - prefix = data[:15] - suffix = data[-8:] - - if encrypted: - payload_len = struct.unpack(">B",data[15:16])[0] # big-endian, unsigned char - version = data[16:19] - checksum = data[19:35] - encrypted_json = data[35:-8] - - json_data = tinytuya.AESCipher(LOCAL_KEY.encode(mock_byte_encoding)).decrypt(encrypted_json) - else: - json_data = data[16:-8].decode(mock_byte_encoding) - - frame_ok = True - if prefix != tinytuya.hex2bin(expected_prefix): - frame_ok = False - elif suffix != tinytuya.hex2bin("000000000000aa55"): - frame_ok = False - elif encrypted: - if payload_len != len(version) + len(checksum) + len(encrypted_json) + len(suffix): - frame_ok = False - elif version != b"3.1": - frame_ok = False - - return json_data, frame_ok - -def mock_send_receive_set_timer(data): - if mock_send_receive_set_timer.call_counter == 0: - ret = 20*chr(0x0) + '{"devId":"DEVICE_ID","dps":{"1":false,"2":0}}' + 8*chr(0x0) - elif mock_send_receive_set_timer.call_counter == 1: - expected = '{"uid":"DEVICE_ID_HERE","devId":"DEVICE_ID_HERE","t":"","dps":{"2":6666}}' - json_data, frame_ok = check_data_frame(data, "000055aa0000000000000007000000") - - if frame_ok and compare_json_strings(json_data, expected, ['t']): - ret = '{"test_result":"SUCCESS"}' - else: - ret = '{"test_result":"FAIL"}' - - ret = ret.encode(mock_byte_encoding) - mock_send_receive_set_timer.call_counter += 1 - return ret - -def mock_send_receive_set_status(data): - expected = '{"dps":{"1":true},"uid":"DEVICE_ID_HERE","t":"1516117564","devId":"DEVICE_ID_HERE"}' - json_data, frame_ok = check_data_frame(data, "000055aa0000000000000007000000") - - if frame_ok and compare_json_strings(json_data, expected, ['t']): - ret = '{"test_result":"SUCCESS"}' - else: - logging.error("json data not the same: {} != {}".format(json_data, expected)) - ret = '{"test_result":"FAIL"}' - - ret = ret.encode(mock_byte_encoding) - return ret - -def mock_send_receive_status(data): - expected = '{"devId":"DEVICE_ID_HERE","gwId":"DEVICE_ID_HERE"}' - json_data, frame_ok = check_data_frame(data, "000055aa000000000000000a000000", False) - - # FIXME dead code block - if frame_ok and compare_json_strings(json_data, expected): - ret = '{"test_result":"SUCCESS"}' - else: - logging.error("json data not the same: {} != {}".format(json_data, expected)) - ret = '{"test_result":"FAIL"}' - - ret = 20*chr(0) + ret + 8*chr(0) - ret = ret.encode(mock_byte_encoding) - return ret - -def mock_send_receive_set_colour(data): - expected = '{"dps":{"2":"colour", "5":"ffffff000000ff"}, "devId":"DEVICE_ID_HERE","uid":"DEVICE_ID_HERE", "t":"1516117564"}' - - json_data, frame_ok = check_data_frame(data, "000055aa0000000000000007000000") - - if frame_ok and compare_json_strings(json_data, expected, ['t']): - ret = '{"test_result":"SUCCESS"}' - else: - logging.error("json data not the same: {} != {}".format(json_data, expected)) - ret = '{"test_result":"FAIL"}' - - ret = ret.encode(mock_byte_encoding) - return ret - -def mock_send_receive_set_white(data): - expected = '{"dps":{"2":"white", "3":255, "4":255}, "devId":"DEVICE_ID_HERE","uid":"DEVICE_ID_HERE", "t":"1516117564"}' - json_data, frame_ok = check_data_frame(data, "000055aa0000000000000007000000") - - if frame_ok and compare_json_strings(json_data, expected, ['t']): - ret = '{"test_result":"SUCCESS"}' - else: - logging.error("json data not the same: {} != {}".format(json_data, expected)) - ret = '{"test_result":"FAIL"}' - - ret = ret.encode(mock_byte_encoding) - return ret + +def get_results_from_mock(d): + result_message_payload = d._send_receive.call_args[0][0] + result_cmd = result_message_payload.cmd + result_payload = json.loads(result_message_payload.payload.decode(mock_byte_encoding)) + result_payload["t"] = "" # clear "t" + + return result_cmd, result_payload + class TestXenonDevice(unittest.TestCase): def test_set_timer(self): + # arrange d = tinytuya.OutletDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY) d.set_version(3.1) - d._send_receive = MagicMock(side_effect=mock_send_receive_set_timer) + mock_response = {"devId":"DEVICE_ID","dps":{"1":False,"2":0}} + d._send_receive = MagicMock(return_value=mock_response) - # Reset call_counter and start test - mock_send_receive_set_timer.call_counter = 0 - result = d.set_timer(6666) - result = result[result.find(b'{'):result.rfind(b'}')+1] - result = result.decode(mock_byte_encoding) # Python 3 (3.5.4 and earlier) workaround to json stdlib "behavior" https://docs.python.org/3/whatsnew/3.6.html#json - result = json.loads(result) + # act + d.set_timer(6666) - # Make sure mock_send_receive_set_timer() has been called twice with correct parameters - self.assertEqual(result['test_result'], "SUCCESS") + # gather results + result_cmd, result_payload = get_results_from_mock(d) + + # expectations + expected_cmd = tinytuya.CONTROL + expected_payload = {"uid":"DEVICE_ID_HERE","devId":"DEVICE_ID_HERE","t":"","dps":{"2":6666}} + + # assert + self.assertEqual(result_cmd, expected_cmd) + self.assertDictEqual(result_payload, expected_payload) def test_set_status(self): + # arrange d = tinytuya.OutletDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY) d.set_version(3.1) - d._send_receive = MagicMock(side_effect=mock_send_receive_set_status) + d._send_receive = MagicMock(return_value={"devId":"DEVICE_ID","dps":{"1":False,"2":0}}) + + # act + d.set_status(True, 1) + + # gather results + result_cmd, result_payload = get_results_from_mock(d) - result = d.set_status(True, 1) - result = result.decode(mock_byte_encoding) # Python 3 (3.5.4 and earlier) workaround to json stdlib "behavior" https://docs.python.org/3/whatsnew/3.6.html#json - result = json.loads(result) + # expectations + expected_cmd = tinytuya.CONTROL + expected_payload = {"dps":{"1":True},"uid":"DEVICE_ID_HERE","t":"","devId":"DEVICE_ID_HERE"} - # Make sure mock_send_receive_set_timer() has been called twice with correct parameters - self.assertEqual(result['test_result'], "SUCCESS") + # assert + self.assertEqual(result_cmd, expected_cmd) + self.assertDictEqual(result_payload, expected_payload) def test_status(self): + # arrange d = tinytuya.OutletDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY) d.set_version(3.1) - d._send_receive = MagicMock(side_effect=mock_send_receive_status) + d._send_receive = MagicMock(return_value={"devId":"DEVICE_ID","dps":{"1":False,"2":0}}) - result = d.status() + # act + d.status() + + # gather results + result_cmd, result_payload = get_results_from_mock(d) + + # expectations + expected_cmd = tinytuya.DP_QUERY + expected_payload = {"devId": "DEVICE_ID_HERE", "gwId": "DEVICE_ID_HERE", "uid": "DEVICE_ID_HERE", "t": ""} + + # assert + self.assertEqual(result_cmd, expected_cmd) + self.assertDictEqual(result_payload, expected_payload) - # Make sure mock_send_receive_set_timer() has been called twice with correct parameters - self.assertEqual(result['test_result'], "SUCCESS") - def test_set_colour(self): + # arrange d = tinytuya.BulbDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY) + d.status = MagicMock(return_value={}) # set_version calls this to figure out which commands it supports d.set_version(3.1) - d._send_receive = MagicMock(side_effect=mock_send_receive_set_colour) + d._send_receive = MagicMock(return_value={"devId":"DEVICE_ID","dps":{"2":"colour", "5":"ffffff000000ff"}}) + + # act + d.set_colour(255,255,255) - result = d.set_colour(255,255,255) - result = result.decode(mock_byte_encoding) - result = json.loads(result) + # gather results + result_cmd, result_payload = get_results_from_mock(d) - self.assertEqual(result['test_result'], "SUCCESS") + # expectations + expected_cmd = tinytuya.CONTROL + # expected_payload = {"dps":{"2":"colour", "5":"ffffff000000ff"}, "devId":"DEVICE_ID_HERE","uid":"DEVICE_ID_HERE", "t": ""} + expected_payload = {"dps":{"21":"colour", "24":"0000000003e8"}, "devId":"DEVICE_ID_HERE","uid":"DEVICE_ID_HERE", "t": ""} + + # assert + self.assertEqual(result_cmd, expected_cmd) + self.assertDictEqual(result_payload, expected_payload) def test_set_white(self): + # arrange d = tinytuya.BulbDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY) + d.status = MagicMock(return_value={}) # set_version calls this to figure out which commands it supports d.set_version(3.1) - d._send_receive = MagicMock(side_effect=mock_send_receive_set_white) + d._send_receive = MagicMock(return_value={"devId":"DEVICE_ID","dps":{"1":False,"2":0}}) + + # act + d.set_white(255, 255) + + # gather results + result_cmd, result_payload = get_results_from_mock(d) + + # expectations + expected_cmd = tinytuya.CONTROL + # expected_payload = {"dps":{"2":"white", "3": 255, "4": 255}, "devId": "DEVICE_ID_HERE","uid": "DEVICE_ID_HERE", "t": ""} + expected_payload = {"dps":{"21":"white", "22": 255, "23": 255}, "devId": "DEVICE_ID_HERE","uid": "DEVICE_ID_HERE", "t": ""} - result = d.set_white(255, 255) - result = result.decode(mock_byte_encoding) - result = json.loads(result) + # assert + self.assertEqual(result_cmd, expected_cmd) + self.assertDictEqual(result_payload, expected_payload) - self.assertEqual(result['test_result'], "SUCCESS") if __name__ == '__main__': unittest.main()