~aleteoryx/muditaos

cdfadba379aec85a359096648b01a5493cc7e3f6 — Bartosz Cichocki 5 years ago 675dc59
[EGD-4510] testing: added pytest tests, updated documentation (#1096)

Add pytest test and functional tests, update documentation

[EGD-4510] rewritten service-desktop tests into pytest manner

[EGD-4510] updated docs, rewritten FT to pytest manner

Co-authored-by: SP2FET <bartosz.cichocki@mudita.com>
35 files changed, 743 insertions(+), 763 deletions(-)

M changelog.md
M test/README.md
M test/harness/harness.py
M test/harness/interface/CDCSerial.py
M test/harness/interface/defs.py
M test/make_a_call.py
A test/pytest/conftest.py
A test/pytest/service-desktop/test_backup.py
A test/pytest/service-desktop/test_calendar.py
A test/pytest/service-desktop/test_calllog.py
A test/pytest/service-desktop/test_contacts.py
A test/pytest/service-desktop/test_device_info.py
A test/pytest/service-desktop/test_factory_reset.py
A test/pytest/service-desktop/test_messages.py
A test/pytest/service-desktop/test_templates.py
A test/pytest/service-desktop/test_update.py
A test/pytest/test_call.py
A test/pytest/test_harness.py
A test/pytest/test_send_message.py
A test/requirements.txt
M test/send_message.py
D test/service-desktop-test/defs.py
D test/service-desktop-test/main.py
D test/service-desktop-test/test_api.py
D test/service-desktop-test/tests/__init__.py
D test/service-desktop-test/tests/backup.py
D test/service-desktop-test/tests/calendarEvents.py
D test/service-desktop-test/tests/calllog.py
D test/service-desktop-test/tests/contacts.py
D test/service-desktop-test/tests/developerMode.py
D test/service-desktop-test/tests/deviceinfo.py
D test/service-desktop-test/tests/factoryReset.py
D test/service-desktop-test/tests/messages.py
D test/service-desktop-test/tests/templates.py
D test/service-desktop-test/tests/update.py
M changelog.md => changelog.md +2 -0
@@ 20,6 20,8 @@
### Other

* `[utils]` Added unit tests for time display.
* `[test]` Rewritten tests and updated test documentation.


### Changed


M test/README.md => test/README.md +128 -49
@@ 1,62 1,141 @@
dumb test harness py
====================
# Test framework

If something is stupid but it works, it's not stupid
Test harness makes use of USB-CDC serial (on RT1051 target) or pseudo-tty (on Linux) to communicate
with `service-desktop` service in the operating system and get data from the internal database and device status.

Simple utility to run key presses, AT commands & wait for response form phone scriptable in python.
Great for repetitive tasks as i.e. phone unlock, or UI tests with multiple key presses.
Both `service-desktop` and functional tests are using [`pyTest`](https://github.com/pytest-dev/pytest/) -
 a testing framework written for Python. 
Convenient usage of pre-defined fixtures and `harness` API enables quick development of further tests.

# running in console
## Test harness API
As a part of the test bundle, test harness is responsible for low-level communication with target or linux simulator.
As for now, it consists of the following methods that can be used during writing tests:

run:
```
python3
from harness.common import conn
* `Harness(port_name)`
 
    constructor used to create `Harness` object and open a serial connection to the operating system
    
* `get_connection()`

   returns a `CDCSerial` object - wrapper for Python's `Serial`
   
* `get_window_name()`

   returns current application(window) name
   
* `with_phone_unlocked(func)`

    decorator allowing to call a function with an unlocked operating system
    
* `open_application(application: str)`

    opens application from main screen (unlocked phone with clock on top)
    
    possible valid arguments: `phone, contacts, messages, music, meditation, settings, tools, alarm, calendar`

* `send_text(text: str)`

    emulates sending text using numpad presses to the text window (eg. SMS)

* `send_number(number: str)`

    emulates sending numbers to the number window (eg. contact number)
    
* `endpoint_request(ep_name: str, met: str, body: dict) -> dict`

    method used to send endpoint request to the service-desktop API and gather requested data from the operating system

    * `ep_name` - valid options: `deviceInfo, update, filesystemUpload, backup, restore, factory, contacts, messages,
     calllog, events, developerMode`
    * `met` - HTTP methods: `get, post, put, del`
    * `body` - JSON payload

#### CDCSerial API
CDCSerial is `Serial` wrapper intended to encapsulate commands and data into communication protocol thus 
allowing to seamlessly transfer data between `Harness` and the operating system. It has the following user-applicable
methods:

# for key press: 0 - 9, second parameter = wait time after key press, here set to '1'
conn.key(0, 1)
# ...
conn.key(9, 1)

# send AT command - '\r' is important - othervise modem will ignore command
conn.gsm("AT?\r") 
* `send_key(key_code, key_type=Keytype.short_press, wait=10)`

# wait for some data
data = conn.read(4)
    emulates sending one keypress as short press (by default) or as a long press (with `Keytype.long_press` argument)
    
    * `key_code` - value of character from `harness.interface.defs.key_codes` array, for numbers `0-9` just ASCII value
    * `wait` - timeout
    
* `send_at(at_command: str, wait=10)`

    send AT command to the Cellular module, returns AT response
    
    * `at_command` - string with AT command, without any terminator
    * `wait` - timeout
    
#### Example
```python
from harness.harness import Harness
from harness.interface.defs import key_codes

port_name = "/dev/ttyACM0"

# init Harness object and open serial port
harness = Harness(port_name)

#get current application name
current_window = harness.get_window_name()

#open messages when phone is unlocked
@harness.with_phone_unlocked()
def do_after_unlock(connection):
    harness.open_application("messages")
    #send joystick down keypress
    connection.send_key(key_codes["down"])
```

# how:
### pyTest running
To execute pyTest test cases make sure that you have installed the `pyTest` package and you have `python3`. It is 
recommended to execute those tests in a virtual environment (eg. in `test` dir):

Frame:
```shell script
pip3 install virtualenv 
virtualenv -p python3 test_env
source test_env/bin/activate
pip3 install -r requirements.txt
```
and then execute the tests (located in `pytest` subdirectory and below):
```shell script
pytest ./pytest --port=/dev/ttyACM2 --timeout=20 
```
<STX>
{
    Type : harness::Enum::Events
    Data : [ <element for selected data type>, ]
}
<ETX>
by default, call and message sending tests are enabled thus you have to provide additional parameters:
```shell script
--phone_number=123456789 --call_duration=30 --sms_text="sms text"
```
which gives full command like this:
```shell script
pytest ./pytest --port=/dev/ttyACM2 --timeout=20  --phone_number=123456789 --call_duration=30 --sms_text="sms text"
```
To run the tests with Linux simulator just pass `simulator` as the port parameter and use `-m` switch with marker negation:
```shell script
pytest ./pytest --port=simulator --timeout=20 -m "not rt1051"
```
`timeout` parameter sets port discovery timeout in seconds (if after this timeout port is not available, pyTest will exit).

After finished test session just call:
```shell script
deactivate
```
to exit the virtual environment.

### sample pyTest test
All tests in the default configuration of pyTest require test files starting with `test_` and test functions also 
starting with `test_`. To use fixture like `harness`, just pass it as an argument to the test function:
```python
test_first_sample_test.py

* Frames start with Frame enum
* Fata is per enum
* All data between `<STX>` abd `<ETX>` is Json
* Binary data should be encoded (base64)
* Partial frames can be added easily

I considered protobuf, tlv and msgpack, protobuf ain't very easy when you want to have Any types inside, msgpack don't have schema.
Json is super easy - json it is

## In app:

* bsp - UART3 added (can be USB, IP whatever really)
    * STX signals application that old data have to be ignored
    * ETX signals app to parse data
* EventManager
    * Informs on start of Eventmanager
    * Informs about application focus change
    * Reads emits fake key releases
* `test/*.py` - test code for handling phone
    * phone_start.py - wait till phone sends its on
    * phone_unlock.py - unlock phone and show desktop
    * switch_app.py - switch to sms app and back waiting for info on application focus change
    * walk presses on `arrow keys` for ~30 min
def test_sample_test(harness):
    body = {}
    ret = harness.endpoint_request("deviceInfo", "get", body)
    assert ret["status"] == 200
   
``` 
All tests written in this manner will be found by pyTest.
#### Notice: all tests that are applicable only to the target should be marked with a `@pytest.mark.rt1051` decorator

M test/harness/harness.py => test/harness/harness.py +8 -3
@@ 13,13 13,18 @@ import random
class Harness:
    connection = None
    is_phone_unlocked = False
    port_name = ''

    def __init__(self, port='/dev/ttyACM2'):
    def __init__(self, port):
        self.port_name = port
        self.connection = serial.CDCSerial(port)

    def get_connection(self):
        return self.connection

    def get_window_name(self):
        return self.connection.get_window()

    def unlock_phone(self):
        if self.connection.is_phone_locked():
            self.connection.send_key(key_codes["enter"])


@@ 31,11 36,11 @@ class Harness:
            log.info("Phone unlocked")
        else:
            log.info("Phone already unlocked")
        self.is_phone_unlocked = True

    def with_phone_unlocked(self, func):
        if not self.is_phone_unlocked:
            self.unlock_phone()
            self.is_phone_unlocked = True

        func(self.connection)



@@ 56,4 61,4 @@ class Harness:
            "uuid": random.randint(1, 32000),
            "body": body
        })
        return ret["body"]
        return ret

M test/harness/interface/CDCSerial.py => test/harness/interface/CDCSerial.py +5 -3
@@ 1,6 1,7 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import time
from random import randrange

import serial
import json


@@ 19,7 20,7 @@ class Keytype(Enum):


class CDCSerial:
    def __init__(self, port_name, timeout=30):
    def __init__(self, port_name, timeout=10):
        self.timeout = timeout
        self.body = ""
        while timeout != 0:


@@ 46,7 47,7 @@ class CDCSerial:
        msg = {
            "endpoint": endpoint["developerMode"],
            "method": method["put"],
            "uuid": 0,
            "uuid": randrange(1,100),
            "body": body
        }
        return msg


@@ 82,9 83,10 @@ class CDCSerial:
        }

        ret = self.write(self.__wrap_message(body), wait)
        print(ret)
        return ret["body"]["ATResponse"]

    def get_window(self):
    def get_window_name(self):
        body = {
            "focus": True
        }

M test/harness/interface/defs.py => test/harness/interface/defs.py +5 -2
@@ 11,7 11,8 @@ endpoint = {
    "contacts": 7,
    "messages": 8,
    "calllog": 9,
    "developerMode": 10
    "events": 10,
    "developerMode": 11
}

method = {


@@ 23,8 24,10 @@ method = {

status = {
    "OK": 200,
    "Accepted": 202,
    "BadRequest": 400,
    "InternalServerError": 500
    "NotAcceptable": 406,
    "InternalServerError": 500,
}

key_codes = {

M test/make_a_call.py => test/make_a_call.py +1 -1
@@ 25,7 25,7 @@ def call(harness, phone_number: str, duration: int):

def get_calllog_count(harness):
    body = {"count": True}
    return harness.endpoint_request("calllog", "get", body)["count"]
    return harness.endpoint_request("calllog", "get", body)["body"]["count"]


def main():

A test/pytest/conftest.py => test/pytest/conftest.py +90 -0
@@ 0,0 1,90 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md

import time

import pytest

import sys
import os.path

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))

from harness import log
from harness.harness import Harness
from harness.interface.error import TestError, Error

simulator_port = 'simulator'


def pytest_addoption(parser):
    parser.addoption("--port", type=str, action="store", required=True)
    parser.addoption("--timeout", type=int, action="store", default=15)
    parser.addoption("--phone_number", type=int, action="store")
    parser.addoption("--call_duration", type=int, action="store", default=30)
    parser.addoption("--sms_text", type=str, action="store", default='')


@pytest.fixture(scope='session')
def phone_number(request):
    phone_number = request.config.option.phone_number
    assert phone_number
    return phone_number


@pytest.fixture(scope='session')
def call_duration(request):
    call_duration = request.config.option.call_duration
    assert call_duration
    return call_duration


@pytest.fixture(scope='session')
def sms_text(request):
    sms_text = request.config.option.sms_text
    assert sms_text != ''
    return sms_text


@pytest.fixture(scope='session')
def harness(request):
    port_name = request.config.option.port
    timeout = request.config.option.timeout

    if port_name is None:
        pytest.exit("no port provided!")
    assert '/dev' in port_name or simulator_port in port_name

    if simulator_port in port_name:
        while timeout != 0:
            try:
                file = open("/tmp/purephone_pts_name", "r")
                break
            except FileNotFoundError as err:
                time.sleep(1)
                timeout = timeout - 1
                print("waiting...")
                if timeout == 0:
                    raise TestError(Error.PORT_FILE_NOT_FOUND)

        port_name = file.readline()
        if port_name.isascii():
            log.debug("found {} entry!".format(port_name))
        else:
            pytest.exit("not a valid sim pts entry!")

    harness = Harness(port_name)
    return harness


@pytest.fixture(scope='session')
def phone_unlocked(harness):
    harness.unlock_phone()
    assert harness.is_phone_unlocked


def pytest_configure(config):
    config.addinivalue_line("markers",
                            "service_desktop_test: mark test if it's related to service-desktop API")
    config.addinivalue_line("markers",
                            "rt1051: mark test if it's target only (eg. calls, messages)")

A test/pytest/service-desktop/test_backup.py => test/pytest/service-desktop/test_backup.py +33 -0
@@ 0,0 1,33 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import pytest
from harness.interface.defs import status


@pytest.mark.service_desktop_test
def test_backup(harness):
    body = {"backupReady": True}
    ret = harness.endpoint_request("backup", "get", body)
    assert ret["status"] == status["OK"]
    assert type(ret["body"]["backupReady"]) is bool

    body = {}
    ret = harness.endpoint_request("backup", "get", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["backupRequest"] is False

    body = {"backupRequest": True}
    ret = harness.endpoint_request("backup", "get", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["backupRequest"] is True

    body = {"backupReady": True}
    ret = harness.endpoint_request("backup", "get", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["backupReady"] is True

    body = {"backupUpload": True}
    ret = harness.endpoint_request("backup", "post", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["backupUpload"] is False


A test/pytest/service-desktop/test_calendar.py => test/pytest/service-desktop/test_calendar.py +63 -0
@@ 0,0 1,63 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import pytest
from harness.interface.defs import status


@pytest.mark.skip("ICS format bug")
@pytest.mark.service_desktop_test
def test_calendar(harness):
    # add event
    event = "BEGIN:VCALENDAR\nBEGIN:VEVENT\nSUMMARY:Testowy\nDTSTART:20200929T123611\nDTEND:20200929T124611\nBEGIN" \
            ":VALARM\nTRIGGER:-P5M\nEND:VALARM\nEND:VEVENT\nEND:VCALENDAR\n "

    body = {"data": event}
    ret = harness.endpoint_request("events", "put", body)
    assert ret["status"] == status["OK"]

    # get all limited events
    body = {"offset": 0, "limit": 1}
    ret = harness.endpoint_request("events", "get", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["data"] == event
    assert ret["body"]["count"] == 1

    uid_index = ret["body"]["data"].index('UID:')
    summary_index = ret["body"]["data"].index('SUMMARY:')
    UID = ret["body"]["data"][uid_index + 4:summary_index]

    assert "SUMMARY:Testowy" in ret["body"]["data"]
    assert "DTSTART:20200929T123611" in ret["body"]["data"]
    assert "DTEND:20200929T124611" in ret["body"]["data"]
    assert ret["body"]["count"] == "1"

    # update event
    event_update = "BEGIN:VCALENDAR\nBEGIN:VEVENT\nUID:" + UID + "\n" + "SUMMARY:Update\nDTSTART:20200928T123611" \
                                                                        "\nDTEND:20200928T124611\nEND:VEVENT\nEND" \
                                                                        ":VCALENDAR\n "
    body = {"data": event_update}
    ret = harness.endpoint_request("events", "post", body)
    assert ret["status"] == status["OK"]

    # get updated event
    body = {"offset": 0, "limit": 1}
    ret = harness.endpoint_request("events", "get", body)
    assert ret["status"] == status["OK"]

    assert ret["body"]["data"] == event_update
    assert ret["body"]["count"] == 1

    assert "SUMMARY:Update" in ret["body"]["data"]
    assert "DTSTART:20200928T123611" in ret["body"]["data"]
    assert "DTEND:20200928T124611" in ret["body"]["data"]
    assert ret["body"]["count"] == "1"

    # remove event
    body = {"UID": UID}
    ret = harness.endpoint_request("events", "del", body)
    assert ret["status"] == status["OK"]

    # check events after remove
    body = {"data": "", "count": "1"}
    ret = harness.endpoint_request("events", "get", body)
    assert ret["status"] == status["OK"]

A test/pytest/service-desktop/test_calllog.py => test/pytest/service-desktop/test_calllog.py +50 -0
@@ 0,0 1,50 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import pytest
from harness.interface.defs import status


@pytest.mark.service_desktop_test
def test_calllog(harness):
    # checking call log count
    body = {"count": True}
    ret = harness.endpoint_request("calllog", "get", body)
    assert ret["status"] == status["OK"]

    count = ret["body"]["count"]
    if count == 0:
        pytest.skip("No calllog entries, skipping")

    # getting the whole call log
    body = {"limit": count}
    ret = harness.endpoint_request("calllog", "get", body)
    assert ret["status"] == status["OK"]

    calllog = ret["body"]
    assert len(calllog)
    contact_id = calllog[0]["contactId"]

    for record in calllog:
        assert record["phoneNumber"].replace('+', '').isnumeric()
        assert record["date"].isnumeric()
        assert record["duration"].isnumeric()

    # getting exact call log
    body = {"contactID": contact_id}
    ret = harness.endpoint_request("calllog", "get", body)
    assert ret["status"] == status["OK"]

    for log in ret["body"]:
        assert log["contactId"] == contact_id

    # remove exact call log
    body = {"id": calllog[0]["id"]}
    ret = harness.endpoint_request("calllog", "del", body)
    assert ret["status"] == status["OK"]

    # getting the count again
    body = {"count": True}
    ret = harness.endpoint_request("calllog", "get", body)
    assert ret["status"] == status["OK"]

    assert count == ret["body"]["count"] + 1

A test/pytest/service-desktop/test_contacts.py => test/pytest/service-desktop/test_contacts.py +89 -0
@@ 0,0 1,89 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import pytest
from harness.interface.defs import status


@pytest.mark.service_desktop_test
def test_contacts(harness):
    # getting the contacts count
    body = {"count": True}
    ret = harness.endpoint_request("contacts", "get", body)
    assert ret["status"] == status["OK"]

    count = ret["body"]["count"]
    if count == 0:
        pytest.skip("No contacts entries, skipping")

    # getting all contacts
    body = {"count": count}
    ret = harness.endpoint_request("contacts", "get", body)
    assert ret["status"] == status["OK"]

    contacts = ret["body"]
    contacts_length = len(contacts)
    assert contacts_length
    assert contacts_length == count

    # adding new contact
    body = {"address": "6 Czeczota St.\n02600 Warsaw",
            "altName": "Testowy",
            "blocked": True,
            "favourite": True,
            "numbers": ["547623521"],
            "priName": "Test"}
    ret = harness.endpoint_request("contacts", "put", body)
    assert ret["status"] == status["OK"]
    contact_id_to_update = ret["body"]["id"]
    assert contact_id_to_update

    # adding new contact without number - should fail with 406
    body = {"address": "6 Czeczota St.\n02600 Warsaw",
            "altName": "Testowy",
            "blocked": True,
            "favourite": True,
            "numbers": [],
            "priName": "Test"}
    ret = harness.endpoint_request("contacts", "put", body)
    assert ret["status"] == status["NotAcceptable"]

    # checking count after adding
    body = {"count": True}
    ret = harness.endpoint_request("contacts", "get", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["count"] == count + 1

    # updating existing contact
    body = {"address": "6 Czeczota St.\n02600 Warsaw",
            "altName": "Testowy2",
            "blocked": True,
            "favourite": True,
            "numbers": ["547623521"],
            "priName": "Test2",
            "id": contact_id_to_update}
    ret = harness.endpoint_request("contacts", "post", body)
    assert ret["status"] == status["OK"]

    # gathering updated contact
    body = {"id": contact_id_to_update}
    ret = harness.endpoint_request("contacts", "get", body)
    contact = {"address": "6 Czeczota St.\n02600 Warsaw",
               "altName": "Testowy2",
               "blocked": True,
               "favourite": True,
               "numbers": ["547623521"],
               "priName": "Test2",
               "id": contact_id_to_update}
    assert ret["body"] == contact

    # removing added contact
    body = {"id": contact_id_to_update}
    ret = harness.endpoint_request("contacts", "del", body)
    assert ret["status"] == status["OK"]

    # verifying count
    body = {"count": True}
    ret = harness.endpoint_request("contacts", "get", body)
    assert ret["status"] == status["OK"]

    assert ret["body"]["count"] == count

A test/pytest/service-desktop/test_device_info.py => test/pytest/service-desktop/test_device_info.py +13 -0
@@ 0,0 1,13 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import pytest
from harness.interface.defs import status


@pytest.mark.service_desktop_test
def test_device_info(harness):
    body = {}
    ret = harness.endpoint_request("deviceInfo", "get", body)
    assert ret["status"] == status["OK"]

    assert 0 < int(ret["body"]["fsFreePercent"]) <= 100

A test/pytest/service-desktop/test_factory_reset.py => test/pytest/service-desktop/test_factory_reset.py +13 -0
@@ 0,0 1,13 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import pytest
from harness.interface.defs import status


@pytest.mark.service_desktop_test
@pytest.mark.skip(reason="This will make a factory reset")
def test_factory_reset(harness):
    body = {"factoryRequest": True}
    ret = harness.endpoint_request("factory", "post", body)
    assert ret["status"] == status["OK"]
    assert ret["body"] == body

A test/pytest/service-desktop/test_messages.py => test/pytest/service-desktop/test_messages.py +57 -0
@@ 0,0 1,57 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md

import pytest
from harness.interface.defs import status


@pytest.mark.service_desktop_test
def test_messages(harness):
    # getting the messages count
    body = {"count": True}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]

    count = ret["body"]["count"]
    if count == 0:
        pytest.skip("No contacts entries, skipping")

    # getting all the messages
    body = {"count": count}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]

    messages = ret["body"]
    messages_count = len(messages)
    assert messages_count == count

    # remove message
    sms_to_remove = messages[0]
    body = {"id": sms_to_remove["id"]}
    ret = harness.endpoint_request("messages", "del", body)
    assert ret["status"] == status["OK"]

    # getting the messages count again
    body = {"count": True}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]

    assert ret["body"]["count"] == count - 1

    # getting messages binded to contactID
    contact_id = 2  # in test dataset this one has some messages
    body = {"contactID": contact_id}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]

    for message in ret["body"]:
        assert message["contactID"] == contact_id

    # getting messages binded to threadID
    thread_id = 1
    body = {"threadID": thread_id, "count": 10}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]

    for message in ret["body"]:
        assert message["threadID"] == thread_id

A test/pytest/service-desktop/test_templates.py => test/pytest/service-desktop/test_templates.py +62 -0
@@ 0,0 1,62 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import pytest
from harness.interface.defs import status


@pytest.mark.service_desktop_test
def test_messages(harness):
    # getting the templates count
    body = {"template": True, "count": True}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]
    count = ret["body"]["count"]

    if count == 0:
        body = {"template": True, "text": "first template"}
        ret = harness.endpoint_request("messages", "put", body)
        assert ret["status"] == status["OK"]

        body = {"template": True, "count": True}
        ret = harness.endpoint_request("messages", "get", body)
        assert ret["status"] == status["OK"]
        count = ret["body"]["count"]
        assert count

    # getting all templates
    body = {"template": True, "count": count}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]
    assert len(ret["body"]) == count

    # adding new template
    body = {"template": True, "text": "test template"}
    ret = harness.endpoint_request("messages", "put", body)
    assert ret["status"] == status["OK"]

    # getting the templates count again
    body = {"template": True, "count": True}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["count"] == count + 1

    # getting template to remove
    body = {"template": True, "count": count + 1}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]

    for template in ret["body"]:
        if template["text"] == "test template":
            id = template["id"]
            break

    # removing template
    body = {"template": True, "id": id}
    ret = harness.endpoint_request("messages", "del", body)
    assert ret["status"] == status["OK"]

    # getting the templates count again
    body = {"template": True, "count": True}
    ret = harness.endpoint_request("messages", "get", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["count"] == count
\ No newline at end of file

A test/pytest/service-desktop/test_update.py => test/pytest/service-desktop/test_update.py +24 -0
@@ 0,0 1,24 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import pytest
from harness.interface.defs import status


@pytest.mark.service_desktop_test
def test_update(harness):
    body = {}
    ret = harness.endpoint_request("update", "get", body)
    assert ret["status"] == status["OK"]

    file_list = ret["body"]["updateFileList"]

    if len(file_list) == 0:
        pytest.skip("No update files")

    for file in file_list:
        assert file["size"] != 0

    body = {"filename": file_list[0]["name"]}
    ret = harness.endpoint_request("update", "post", body)
    assert ret["status"] == status["OK"]
    assert ret["body"]["updateReady"] is True

A test/pytest/test_call.py => test/pytest/test_call.py +28 -0
@@ 0,0 1,28 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import time
import pytest

from harness.interface.defs import key_codes


def get_calllog_count(harness):
    body = {"count": True}
    return harness.endpoint_request("calllog", "get", body)["body"]["count"]


@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_call(harness, phone_number, call_duration):
    count_before = get_calllog_count(harness)
    # enter number
    harness.send_number(str(phone_number))
    # call
    harness.connection.send_key(key_codes["fnLeft"])
    time.sleep(call_duration)
    # hang up
    harness.connection.send_key(key_codes["fnRight"])
    count_after = get_calllog_count(harness)

    assert count_before + 1 == count_after
    time.sleep(2)  # needed to restore after call

A test/pytest/test_harness.py => test/pytest/test_harness.py +13 -0
@@ 0,0 1,13 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md

import pytest


@pytest.mark.skip("not working ;/")
@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_send_AT(harness):
    ret = harness.connection.send_at("AT")
    print(ret)
    assert "OK" in ret

A test/pytest/test_send_message.py => test/pytest/test_send_message.py +46 -0
@@ 0,0 1,46 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import time
import pytest

from harness.interface.defs import key_codes


def get_message_by_text(harness, message: str):
    body = {"messageBody": message}
    return harness.endpoint_request("messages", "get", body)["body"]


@pytest.mark.rt1051
@pytest.mark.usefixtures("phone_unlocked")
def test_send_message(harness, phone_number, sms_text):
    messages = get_message_by_text(harness, sms_text.upper())

    # enter menu
    harness.connection.send_key(key_codes["enter"])
    harness.open_application("messages")
    if harness.connection.get_window_name() != "ApplicationMessages":
        time.sleep(2)
        assert harness.connection.get_window_name() == "ApplicationMessages"

    # create new message
    harness.connection.send_key(key_codes["left"])
    # enter phone number
    harness.send_number(str(phone_number))
    # move down to message body
    harness.connection.send_key(key_codes["down"])
    # write a message
    harness.send_text(sms_text)
    # send
    harness.connection.send_key(key_codes["enter"])

    # go back to main screen
    for i in range(3):
        harness.connection.send_key(key_codes["fnRight"])

    time.sleep(2)
    new_messages = get_message_by_text(harness, sms_text.upper())

    diff = [i for i in messages + new_messages if i not in messages or i not in new_messages]
    assert len(diff) == 1
    assert diff[0]["type"] == 0x08

A test/requirements.txt => test/requirements.txt +10 -0
@@ 0,0 1,10 @@
attrs==20.3.0
iniconfig==1.1.1
packaging==20.4
pluggy==0.13.1
py==1.9.0
pyparsing==2.4.7
pyserial==3.5
pytest==6.1.2
six==1.15.0
toml==0.10.2

M test/send_message.py => test/send_message.py +3 -3
@@ 17,9 17,9 @@ def send_message(harness, phone_number: str, message: str):
        # enter menu
        connection.send_key(key_codes["enter"])
        harness.open_application("messages")
        if harness.connection.get_window() != "ApplicationMessages":
        if harness.connection.get_window_name() != "ApplicationMessages":
            time.sleep(2)
            if harness.connection.get_window() != "ApplicationMessages":
            if harness.connection.get_window_name() != "ApplicationMessages":
                print("Application didn't switch, exiting...")
                exit(1)



@@ 37,7 37,7 @@ def send_message(harness, phone_number: str, message: str):

def get_message_by_text(harness, message: str):
    body = {"messageBody": message}
    return harness.endpoint_request("messages", "get", body)
    return harness.endpoint_request("messages", "get", body)["body"]


def main():

D test/service-desktop-test/defs.py => test/service-desktop-test/defs.py +0 -27
@@ 1,27 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md

endpoint = {
    "deviceInfo": 1,
    "update": 2,
    "backup": 3,
    "restore": 4,
    "factory": 5,
    "contacts": 6,
    "messages": 7,
    "calllog": 8,
    "events": 10
}

method = {
    "get": 1,
    "post": 2,
    "put": 3,
    "del": 4
}

status = {
    "OK": 200,
    "BadRequest": 400,
    "InternalServerError": 500
}

D test/service-desktop-test/main.py => test/service-desktop-test/main.py +0 -45
@@ 1,45 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
import sys
from tests.messages import *
from tests.templates import *
from tests.contacts import *
from tests.update import *
from tests.deviceinfo import *
from tests.factoryReset import *
from tests.backup import *
from tests.calllog import *
from tests.calendarEvents import *
from termcolor import colored

from harness.harness import Harness


def main():
    if len(sys.argv) == 1 or sys.argv[1] is None:
        print('Please pass port name as the parameter: python main.py /dev/ttyACM0 ')
        exit(1)

    test_harness = Harness(sys.argv[1])
    serial = test_harness.get_connection()
    final_result = True
    failed_tests = []
    for test_instance in (DeviceInfoTest(serial), calendarEventsTest(serial), UpdateTest(serial), BackupTest(serial), MessageTest(serial),
                          MessageTemplateTest(serial), ContactTest(serial), CalllogTest(serial),
                          FactoryResetTest(serial)):
        test_name = type(test_instance).__name__
        result = test_instance.run()
        if result == False:
            failed_tests.append(test_name)
        final_result = final_result and result
        print(
            f'[{test_name}] {colored("PASS", "green") if result else colored("FAIL", "red")}')

    print("\nFinal result:", colored("PASS", "green") if final_result else colored("FAIL", "red"))
    print("Failed tests:")
    for test in failed_tests:
        print(colored(test, "red"))


if __name__ == '__main__':
    main()

D test/service-desktop-test/test_api.py => test/service-desktop-test/test_api.py +0 -45
@@ 1,45 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md


import json
import random


class Test:
    def __init__(self, serial, payload, expected_result):
        self.result = ""
        self.serial = serial
        self.message = build_message(payload)
        self.expected_result = expected_result

    def execute(self):
        self.serial.write(self.message.encode())
        header = self.serial.read(10).decode()
        payload_length = int(header[1:])
        result = self.serial.read(payload_length).decode()
        # print("result full:" + result)
        self.result = json.loads(result)
        return self.result == self.expected_result, self.result["body"]


def build_message(json_data):
    json_dump = json.dumps(json_data)
    return "#%09d%s" % (len(json_dump), json_dump)


def prepare_message(endpoint, method, status, msg_body, result_body={}):
    uuid = random.randint(1, 32768)
    msg = {
        "endpoint": endpoint,
        "method": method,
        "uuid": uuid,
        "body": msg_body
    }
    msg_result = {
        "endpoint": endpoint,
        "status": status,
        "uuid": str(uuid),
        "body": result_body
    }
    return msg, msg_result

D test/service-desktop-test/tests/__init__.py => test/service-desktop-test/tests/__init__.py +0 -9
@@ 1,9 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md

import sys, os, inspect
currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
parentdir = os.path.dirname(currentdir)
parent2dir = os.path.dirname(parentdir)
sys.path.insert(0, parentdir)
sys.path.insert(0, parent2dir)

D test/service-desktop-test/tests/backup.py => test/service-desktop-test/tests/backup.py +0 -52
@@ 1,52 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
from harness.interface.defs import endpoint, method, status

from test_api import *


class BackupTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):

        msg, result_msg = prepare_message(endpoint["backup"], method["get"], status["OK"],
                                          {"backupReady": True}, {"backupReady": False})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            if result["backupReady"] == True:
                print("backup already ready!")
            else:
                return False

        msg, result_msg = prepare_message(endpoint["backup"], method["get"], status["OK"],
                                          {}, {"backupRequest": False})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        msg, result_msg = prepare_message(endpoint["backup"], method["get"], status["OK"],
                                          {"backupRequest": True}, {"backupRequest": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        msg, result_msg = prepare_message(endpoint["backup"], method["get"], status["OK"],
                                          {"backupReady": True}, {"backupReady": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        msg, result_msg = prepare_message(endpoint["backup"], method["post"], status["OK"],
                                          {"backupUpload": True}, {"backupUpload": False})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        return True

D test/service-desktop-test/tests/calendarEvents.py => test/service-desktop-test/tests/calendarEvents.py +0 -101
@@ 1,101 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md

from harness.interface.defs import endpoint, method, status

from test_api import *
from defs import *


class calendarEventsTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):
        # add event
        event = "BEGIN:VCALENDAR\nBEGIN:VEVENT\nSUMMARY:Testowy\nDTSTART:20200929T123611\nDTEND:20200929T124611\nBEGIN:VALARM\nTRIGGER:-P5M\nEND:VALARM\nEND:VEVENT\nEND:VCALENDAR\n"

        msg, result_msg = prepare_message(endpoint["events"], method["put"], status["OK"],
                                          {"data": event}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        #get all limited events
        msg, result_msg = prepare_message(endpoint["events"], method["get"], status["OK"],
                                          {"offset": 0,
                                           "limit":  1},
                                          {"data":event,
                                           "count": "1"})

        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        print(result)
        uid_index = result["data"].index('UID:')
        summary_index = result["data"].index('SUMMARY:')
        UID = result["data"][uid_index+4:summary_index]
        print(UID)

        if not "SUMMARY:Testowy" in result["data"]:
            return False

        if not "DTSTART:20200929T123611" in result["data"]:
            return False

        if not "DTEND:20200929T124611" in result["data"]:
            return False

        if result["count"] != "1":
            return False

        eventUpdate = "BEGIN:VCALENDAR\nBEGIN:VEVENT\nUID:"+UID+"\n"+"SUMMARY:Update\nDTSTART:20200928T123611\nDTEND:20200928T124611\nEND:VEVENT\nEND:VCALENDAR\n"
        print(eventUpdate)
        # # update event
        msg, result_msg = prepare_message(endpoint["events"], method["post"], status["OK"],
                                          {"data":eventUpdate}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        # check updated event
        msg, result_msg = prepare_message(endpoint["events"], method["get"], status["OK"],
                                          {"offset": 0,
                                           "limit":  1},
                                          {"data":eventUpdate,
                                           "count": "1"})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        print(result)

        if not "SUMMARY:Update" in result["data"]:
            return False

        if not "DTSTART:20200928T123611" in result["data"]:
            return False

        if not "DTEND:20200928T124611" in result["data"]:
            return False

        if result["count"] != "1":
            return False

        #remove event
        msg, result_msg = prepare_message(endpoint["events"], method["del"], status["OK"],
                                          {"UID": UID}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        # check events after remove
        msg, result_msg = prepare_message(endpoint["events"], method["get"], status["OK"],None,
                                          {"data":"",
                                           "count": "1"})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        return True

D test/service-desktop-test/tests/calllog.py => test/service-desktop-test/tests/calllog.py +0 -59
@@ 1,59 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
from harness.interface.defs import endpoint, method, status

from test_api import *


class CalllogTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):

        msg, result_msg = prepare_message(endpoint["calllog"], method["get"], status["OK"], {"count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        count = result["count"]
        if count == 0:
            print("count = 0!")
            return False

        msg, result_msg = prepare_message(endpoint["calllog"], method["get"], status["OK"], {"limit": count})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        contactId = result[0]["contactId"]
        if len(result) != count:
            print("record length mismatch!")
            return False

        for record in result:
            if record["phoneNumber"].isnumeric() == False:
                return False
            if record["date"].isnumeric() == False:
                return False
            if record["duration"].isnumeric() == False:
                return False

        msg, result_msg = prepare_message(endpoint["calllog"], method["get"], status["OK"], {"contactID": contactId})
        test = Test(self.serial, msg, result_msg)
        ret, calllog_result = test.execute()
        for calllog in calllog_result:
            if calllog["contactId"] != contactId:
                print("contactID mismatch!")
                return False

        msg, result_msg = prepare_message(endpoint["calllog"], method["del"], status["OK"], {"id": result[0]["id"]},None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret != True:
            return False

        msg, result_msg = prepare_message(endpoint["calllog"], method["get"], status["OK"], {"count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if count != result["count"]+1:
            print("record count after delete mismatch!")
            return False

        return True

D test/service-desktop-test/tests/contacts.py => test/service-desktop-test/tests/contacts.py +0 -113
@@ 1,113 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
from harness.interface.defs import endpoint, method, status

from test_api import *


class ContactTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):
        # get contacts count
        msg, result_msg = prepare_message(endpoint["contacts"], method["get"], status["OK"],
                                          {"count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        count = result['count']
        if count == 0:
            print("count = 0!")
            return False

        # get all contacts
        msg, result_msg = prepare_message(endpoint["contacts"], method["get"], status["OK"],
                                          {"count": count})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        records_length = len(result)
        contact_to_update = result[0]
        if records_length != count:
            print("received record count mismatch!")
            return False

        # add contact
        msg, result_msg = prepare_message(endpoint["contacts"], method["put"], status["OK"],
                                          {"address": "6 Czeczota St.\n02600 Warsaw",
                                           "altName": "Testowy",
                                           "blocked": True,
                                           "favourite": True,
                                           "numbers": ["547623521"],
                                           "priName": "Test"}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        # again check contacts count
        msg, result_msg = prepare_message(endpoint["contacts"], method["get"], status["OK"],
                                          {"count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if result["count"] != count + 1:
            print("record count unchanged after add!")
            return False

        # update contact
        msg, result_msg = prepare_message(endpoint["contacts"], method["post"], status["OK"],
                                          {"address": "6 Czeczota St.\n02600 Warsaw",
                                           "altName": "Testowy2",
                                           "blocked": True,
                                           "favourite": True,
                                           "numbers": ["547623521"],
                                           "priName": "Test2",
                                           "id": contact_to_update["id"]}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        # check updated contact
        msg, result_msg = prepare_message(endpoint["contacts"], method["get"], status["OK"],
                                          {"id": contact_to_update["id"]})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        contact = {"address": "6 Czeczota St.\n02600 Warsaw",
                   "altName": "Testowy2",
                   "blocked": True,
                   "favourite": True,
                   "numbers": ["547623521"],
                   "priName": "Test2",
                   "id": contact_to_update["id"]}
        if result != contact:
            print("updated record mismatch!")
            return False

        # get contact to remove
        msg, result_msg = prepare_message(endpoint["contacts"], method["get"], status["OK"],
                                          {"count": count + 1})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        for contact in result:
            if contact["priName"] == "Test" and contact["altName"] == "Testowy":
                id = contact["id"]
                break

        # remove contact
        msg, result_msg = prepare_message(endpoint["contacts"], method["del"], status["OK"],
                                          {"id": id}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        # again check contacts count
        msg, result_msg = prepare_message(endpoint["contacts"], method["get"], status["OK"],
                                          {"count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if result["count"] != count:
            print("record count unchanged after remove!")
            return False

        return True

D test/service-desktop-test/tests/developerMode.py => test/service-desktop-test/tests/developerMode.py +0 -31
@@ 1,31 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
from harness.interface.defs import endpoint, method, status

from test_api import *
import time


def sendKey(key_code):
    msg, result_msg = prepare_message(endpoint["developerMode"], method["put"], status["OK"],
                                      {"keyPressed": key_code}, None)
    test = Test(self.serial, msg, result_msg)
    test.execute()


class DeveloperModeTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):
        # unlock the phone
        sendKey(key_codes["enter"])
        time.sleep(1)
        sendKey(key_codes["#"])
        time.sleep(1)

        for i in range(4):
            sendKey(3)
            time.sleep(1.5)

        return True

D test/service-desktop-test/tests/deviceinfo.py => test/service-desktop-test/tests/deviceinfo.py +0 -24
@@ 1,24 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md

from harness.interface.defs import endpoint, method, status

from test_api import *


class DeviceInfoTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):

        msg, result_msg = prepare_message(endpoint["deviceInfo"], method["get"], status["OK"], {})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()

        if int(result["fsFreePercent"]) == 0 or int(result["fsFreePercent"]) > 100:
            return False
        if int(result["batteryState"]) != 1:
            return False

        return True

D test/service-desktop-test/tests/factoryReset.py => test/service-desktop-test/tests/factoryReset.py +0 -17
@@ 1,17 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
from harness.interface.defs import endpoint, method, status

from test_api import *


class FactoryResetTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):
        msg, result_msg = prepare_message(endpoint["factory"], method["post"], status["OK"], {"factoryRequest": True},
                                          {'factoryRequest': True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        return ret

D test/service-desktop-test/tests/messages.py => test/service-desktop-test/tests/messages.py +0 -69
@@ 1,69 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md

from harness.interface.defs import endpoint, method, status
from test_api import *


class MessageTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):
        # get messages count
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"], {"count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        count = result['count']
        if count == 0:
            print("count = 0!")
            return False

        # get all messages
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"], {"count": count})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        records_length = len(result)
        if records_length != count:
            print("received record count mismatch!")
            return False

        # remove message
        sms_to_remove = result[0]
        msg, result_msg = prepare_message(endpoint["messages"], method["del"], status["OK"],
                                          {"id": sms_to_remove["id"]}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return ret

        # check if the message was removed
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"], {"count": count})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        for message in result:
            if message["id"] == sms_to_remove["id"]:
                print("contact still exists!")
                return False

        # get messages binded to contactID
        contact_id = 2
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"], {"contactID": contact_id})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        for message in result:
            if message["contactID"] != contact_id:
                print("wrong contactID!")
                return False

        thread_id = 1
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"],
                                          {"threadID": thread_id, "count": 10})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        for message in result:
            if message["threadID"] != thread_id:
                print("wrong threadID!")
                return False

        return True

D test/service-desktop-test/tests/templates.py => test/service-desktop-test/tests/templates.py +0 -77
@@ 1,77 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
from harness.interface.defs import endpoint, method, status

from test_api import *


class MessageTemplateTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):
        # get templates count
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"],
                                          {"template": True, "count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        count = result['count']
        if count == 0:
            print("count = 0!")
            return False

        # get all templates
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"],
                                          {"template": True, "count": count})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        records_length = len(result)
        if records_length != count:
            print("received record count mismatch!")
            return False

        # add template
        msg, result_msg = prepare_message(endpoint["messages"], method["put"], status["OK"],
                                          {"template": True, "text": "test template"}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        # again check templates count
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"],
                                          {"template": True, "count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if result["count"] != count + 1:
            print("record count unchanged after add!")
            return False

        # get template to remove
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"],
                                          {"template": True, "count": count + 1})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        for template in result:
            if template["text"] == "test template":
                id = template["id"]
                break

        # remove template
        msg, result_msg = prepare_message(endpoint["messages"], method["del"], status["OK"],
                                          {"template": True, "id": id}, None)
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if ret == False:
            return False

        # again check templates count
        msg, result_msg = prepare_message(endpoint["messages"], method["get"], status["OK"],
                                          {"template": True, "count": True})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if result["count"] != count:
            print("record count unchanged after remove!")
            return False

        return True

D test/service-desktop-test/tests/update.py => test/service-desktop-test/tests/update.py +0 -33
@@ 1,33 0,0 @@
# Copyright (c) 2017-2020, Mudita Sp. z.o.o. All rights reserved.
# For licensing, see https://github.com/mudita/MuditaOS/LICENSE.md
from harness.interface.defs import endpoint, method, status

from test_api import *


class UpdateTest:
    def __init__(self, serial):
        self.serial = serial.get_serial()

    def run(self):
        # perform update endpoint test
        msg, result_msg = prepare_message(endpoint["update"], method["get"], status["OK"], {})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        filename = ""
        if len(result) > 0:
            filename = result

        if len(filename) > 1:
            filename = filename[0]
        msg, result_msg = prepare_message(endpoint["update"], method["post"], status["OK"], {"filename": filename})
        test = Test(self.serial, msg, result_msg)
        ret, result = test.execute()
        if filename != "":
            if result["updateReady"] == True:
                return True
        else:
            if result["updateReady"] != False:
                return False

        return True