Load standard values in server in testcases #2 (#261)

* added small int support

* change order

* spaces between functions

* changed functions small_int to usint (unsigned small int)

* change in db_row get value y set value changed comparation from sint to usint (0-255)

* added to db_row get_value and set_value comparation to sint (-128 to 127)

* added support for SMALL INT (-127 to 128)

* changes in test_util in bytearray, added numbers to ocuppe all bytes of the data structure in DTL

* test passed | changed _bytearray in test to match the lenght for the test

* added doc

* change space

* arreglos de espacios en blanco

* added s7areaDB  for default

* added default to db and db_row -> area=S7AreaDB

* same

* added read area | write area --- TODO tests

* correction to the db1['test'].read() -> added client

* read area debe corregirse

* correction to client.read_area of function write in class db_row

* corrections to def write in class db_row

* Small changes

- Types.py:
 * Added enum class to handle areas
- Utils:
 * Added some missing types
 * Added area parameter to the DB class, this way you could handle the inputs, outputs, or marks as a DB for reading it TODO: add tests
- Common.py:
   * Missing types
- Client.py:
 * Added an optional init parameter for the lib location, this way people who can't instal properly the library, they can specify the path to the .dll file.

* Types.py readd S7Areas as alone values

* Test fixes 1

* Typing errors fix 2

* White spaces fixes

* White spaces

* Remove enum Areas, so we stick to the ADict class

* Remove it TODOs in util.py

* Utils type export fix

* _type parameter switch to type_

* Fix typing for areas Adict

* Migration to Enum for Areas and WordLen

* switch assert date

* whitespaces

* Fix typing errors in util file

* ignore typing

* fix test

* whitespaces

* Added find locally snap7.dll
Fix logo errors

* fix assert error in date

* types.py:
* remove it string multiline methods added-

common.py
* Remove it type ignore from ctypes import

* Added common test and change common find_locally from using os to pathlib

* Pathlib

* Path lib fix to find in the current working directory

* prefill server areas for testing

* como crear un arreglo desde un buffer

* ...

* test mainloop prefill working until  double int

* Working filled values for:
- DWORD
- WORD
- STRING
- DINT
- INT
- USINT
- SINT
- BOOL

* fix pycodestyle

* Fixed test_read_real adding 0.00001% of almost equal assert

* Server and test changes

- Mainloop init standar values move to it's own function
- Change how binary is written
- Doc added?
- Move each datatype to a round decimal place

* fix pycodestyle

* added test_mainloop to the makefile

* also run tests on github actions

Co-authored-by: Lautaro Dapino <[email protected]>

Author: gijzelaerr

.github/workflows/pytest.yml .github/workflows/linux.yml.github/workflows/pytest.yml renamed to .github/workflows/linux.yml

@@ -32,5 +32,5 @@ jobs:
       - name: Run pytest
         run: |
           which pytest
-          pytest test/test_server.py test/test_client.py test/test_util.py
+          pytest test/test_server.py test/test_client.py test/test_util.py test/test_mainloop.py
           sudo /opt/hostedtoolcache/Python/${{ matrix.python-version }}*/x64/bin/pytest test/test_partner.py

.github/workflows/osx.yml

@@ -18,7 +18,7 @@ jobs:
         run: python3 -m pip install .[test]
       - name: Run pytest
         run: |
-          pytest test/test_server.py test/test_client.py test/test_util.py
+          pytest test/test_server.py test/test_client.py test/test_util.py test/test_mainloop.py
           sudo pytest test/test_partner.py
 
 

.github/workflows/windows.yml

@@ -24,4 +24,4 @@ jobs:
       - name: Install python libraries
         run: python.exe -m pip install .[test]
       - name: Run tests
-        run: pytest.exe  test/test_client.py test/test_util.py test/test_partner.py
+        run: pytest.exe  test/test_client.py test/test_util.py test/test_partner.py test/test_mainloop.py

Makefile

@@ -33,7 +33,7 @@ mypy: venv/bin/pytest
 	venv/bin/mypy snap7 test
 
 test: venv/bin/pytest
-	venv/bin/pytest test/test_server.py test/test_client.py test/test_util.py
+	venv/bin/pytest test/test_server.py test/test_client.py test/test_util.py test/test_mainloop.py
 	sudo venv/bin/pytest test/test_partner.py  # run this as last to prevent pytest cache dir creates as root
 
 clean:

snap7/server.py

@@ -3,6 +3,7 @@
 """
 import logging
 import re
+import struct
 import time
 import ctypes
 from typing import Tuple, Optional, Callable, Any
@@ -290,7 +291,7 @@ def clear_events(self) -> int:
         return self.library.Srv_ClearEvents(self.pointer)
 
 
-def mainloop(tcpport: int = 1102):
+def mainloop(tcpport: int = 1102, init_standard_values: bool = False):
     server = snap7.server.Server()
     size = 100
     DBdata = (snap7.types.wordlen_to_ctypes[snap7.types.WordLen.Byte.value] * size)()
@@ -301,6 +302,13 @@ def mainloop(tcpport: int = 1102):
     server.register_area(snap7.types.srvAreaPA, 1, PAdata)
     server.register_area(snap7.types.srvAreaTM, 1, TMdata)
     server.register_area(snap7.types.srvAreaCT, 1, CTdata)
+
+    if init_standard_values:
+        ba = _init_standard_values()
+        DBdata = snap7.types.wordlen_to_ctypes[snap7.types.WordLen.Byte.value] * len(ba)
+        DBdata = DBdata.from_buffer(ba)
+        server.register_area(snap7.types.srvAreaDB, 0, DBdata)
+
     server.start(tcpport=tcpport)
     while True:
         while True:
@@ -310,3 +318,136 @@ def mainloop(tcpport: int = 1102):
             else:
                 break
         time.sleep(1)
+
+
+def _init_standard_values() -> bytearray:
+    ''' Standard values
+    * Boolean
+    BYTE    BIT     VALUE
+    0       0       True
+    0       1       False
+    0       2       True
+    0       3       False
+    0       4       True
+    0       5       False
+    0       6       True
+    0       7       False
+
+    * Small int
+    BYTE    VALUE
+    10      -128
+    11      0
+    12      100
+    13      127
+
+    * Unsigned small int
+    BYTE    VALUE
+    20      0
+    21      255
+
+    * Int
+    BYTE    VALUE
+    30      -32768
+    32      -1234
+    34      0
+    36      1234
+    38      32767
+
+    * Double int
+    BYTE    VALUE
+    40      -2147483648
+    44      -32768
+    48      0
+    52      32767
+    56      2147483647
+
+    * Real
+    BYTE    VALUE
+    60      -3.402823e38
+    64      -3.402823e12
+    68      -175494351e-38
+    72      -1.175494351e-12
+    76      0.0
+    80      1.175494351e-38
+    84      1.175494351e-12
+    88      3.402823466e12
+    92      3.402823466e38
+
+    * String
+    BYTE    VALUE
+    100     254|37|the brown fox jumps over the lazy dog
+
+    * Word
+    BYTE    VALUE
+    400     \x00\x00
+    404     \x12\x34
+    408     \xAB\xCD
+    412     \xFF\xFF
+
+    * Double Word
+    BYTE    VALUE
+    500     \x00\x00\x00\x00
+    508     \x12\x34\x56\x78
+    516     \x12\x34\xAB\xCD
+    524     \xFF\xFF\xFF\xFF
+    '''
+
+    ba = bytearray(1000)
+    # 1. Bool 1 byte
+    ba[0] = 0b10101010
+
+    # 2. Small int 1 byte
+    ba[10:10 + 1] = struct.pack(">b", -128)
+    ba[11:11 + 1] = struct.pack(">b", 0)
+    ba[12:12 + 1] = struct.pack(">b", 100)
+    ba[13:13 + 1] = struct.pack(">b", 127)
+
+    # 3. Unsigned small int 1 byte
+    ba[20:20 + 1] = struct.pack("B", 0)
+    ba[21:21 + 1] = struct.pack("B", 255)
+
+    # 4. Int 2 bytes
+    ba[30:30 + 2] = struct.pack(">h", -32768)
+    ba[32:32 + 2] = struct.pack(">h", -1234)
+    ba[34:34 + 2] = struct.pack(">h", 0)
+    ba[36:36 + 2] = struct.pack(">h", 1234)
+    ba[38:38 + 2] = struct.pack(">h", 32767)
+
+    # 5. DInt 4 bytes
+    ba[40:40 + 4] = struct.pack(">i", -2147483648)
+    ba[44:44 + 4] = struct.pack(">i", -32768)
+    ba[48:48 + 4] = struct.pack(">i", 0)
+    ba[52:52 + 4] = struct.pack(">i", 32767)
+    ba[56:56 + 4] = struct.pack(">i", 2147483647)
+
+    # 6. Real 4 bytes
+    ba[60:60 + 4] = struct.pack(">f", -3.402823e38)
+    ba[64:64 + 4] = struct.pack(">f", -3.402823e12)
+    ba[68:68 + 4] = struct.pack(">f", -175494351e-38)
+    ba[72:72 + 4] = struct.pack(">f", -1.175494351e-12)
+    ba[76:76 + 4] = struct.pack(">f", 0.0)
+    ba[80:80 + 4] = struct.pack(">f", 1.175494351e-38)
+    ba[84:84 + 4] = struct.pack(">f", 1.175494351e-12)
+    ba[88:88 + 4] = struct.pack(">f", 3.402823466e12)
+    ba[92:92 + 4] = struct.pack(">f", 3.402823466e38)
+
+    # 7. String 1 byte per char
+    string = "the brown fox jumps over the lazy dog"  # len = 37
+    ba[100] = 254
+    ba[101] = len(string)
+    for letter, i in zip(string, range(102, 102 + len(string) + 1)):
+        ba[i] = ord(letter)
+
+    # 8. WORD 4 bytes
+    ba[400:400 + 4] = b"\x00\x00"
+    ba[404:404 + 4] = b"\x12\x34"
+    ba[408:408 + 4] = b"\xAB\xCD"
+    ba[412:412 + 4] = b"\xFF\xFF"
+
+    # # 9 DWORD 8 bytes
+    ba[500:500 + 8] = b"\x00\x00\x00\x00"
+    ba[508:508 + 8] = b"\x12\x34\x56\x78"
+    ba[516:516 + 8] = b"\x12\x34\xAB\xCD"
+    ba[524:524 + 8] = b"\xFF\xFF\xFF\xFF"
+
+    return ba

test/test_mainloop.py

@@ -0,0 +1,139 @@
+import ctypes
+import logging
+from multiprocessing.context import Process
+from os import get_inheritable
+import time
+import unittest
+from unittest import mock
+
+import snap7.error
+import snap7.server
+import snap7.util
+from snap7.util import get_bool, get_dint, get_dword, get_int, get_real, get_sint, get_string, get_usint, get_word
+from snap7.client import Client
+import snap7.types
+
+logging.basicConfig(level=logging.WARNING)
+
+ip = '127.0.0.1'
+tcpport = 1102
+db_number = 1
+rack = 1
+slot = 1
+
+
+class TestServer(unittest.TestCase):
+
+    process = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.process = Process(target=snap7.server.mainloop, args=[tcpport, True])
+        cls.process.start()
+        time.sleep(2)  # wait for server to start
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.process.terminate()
+        cls.process.join(1)
+        if cls.process.is_alive():
+            cls.process.kill()
+
+    def setUp(self):
+        self.client: Client = snap7.client.Client()
+        self.client.connect(ip, rack, slot, tcpport)
+
+    def tearDown(self):
+        self.client.disconnect()
+        self.client.destroy()
+
+    @unittest.skip("TODO: only first test used")
+    def test_read_prefill_db(self):
+        data = self.client.db_read(0, 0, 7)
+        boolean = snap7.util.get_bool(data, 0, 0)
+        print(data)
+        self.assertEqual(boolean, True)
+        integer = snap7.util.get_int(data, 1)
+        self.assertEqual(integer, 128)
+        real = snap7.util.get_real(data, 3)
+        self.assertEqual(real, -128)
+
+    def test_read_booleans(self):
+        data = self.client.db_read(0, 0, 1)
+        self.assertEqual(False, get_bool(data, 0, 0))
+        self.assertEqual(True, get_bool(data, 0, 1))
+        self.assertEqual(False, get_bool(data, 0, 2))
+        self.assertEqual(True, get_bool(data, 0, 3))
+        self.assertEqual(False, get_bool(data, 0, 4))
+        self.assertEqual(True, get_bool(data, 0, 5))
+        self.assertEqual(False, get_bool(data, 0, 6))
+        self.assertEqual(True, get_bool(data, 0, 7))
+
+    def test_read_small_int(self):
+        data = self.client.db_read(0, 10, 4)
+        value_1 = get_sint(data, 0)
+        value_2 = get_sint(data, 1)
+        value_3 = get_sint(data, 2)
+        value_4 = get_sint(data, 3)
+        self.assertEqual(value_1, -128)
+        self.assertEqual(value_2, 0)
+        self.assertEqual(value_3, 100)
+        self.assertEqual(value_4, 127)
+
+    def test_read_unsigned_small_int(self):
+        data = self.client.db_read(0, 20, 2)
+        self.assertEqual(get_usint(data, 0), 0)
+        self.assertEqual(get_usint(data, 1), 255)
+
+    def test_read_int(self):
+        data = self.client.db_read(0, 30, 10)
+        self.assertEqual(get_int(data, 0), -32768)
+        self.assertEqual(get_int(data, 2), -1234)
+        self.assertEqual(get_int(data, 4), 0)
+        self.assertEqual(get_int(data, 6), 1234)
+        self.assertEqual(get_int(data, 8), 32767)
+
+    def test_read_double_int(self):
+        data = self.client.db_read(0, 40, 4 * 5)
+        self.assertEqual(get_dint(data, 0), -2147483648)
+        self.assertEqual(get_dint(data, 4), -32768)
+        self.assertEqual(get_dint(data, 8), 0)
+        self.assertEqual(get_dint(data, 12), 32767)
+        self.assertEqual(get_dint(data, 16), 2147483647)
+
+    def test_read_real(self):
+        data = self.client.db_read(0, 60, 4 * 9)
+        self.assertAlmostEqual(get_real(data, 0), -3.402823e38, delta=-3.402823e38 * -0.0000001)
+        self.assertAlmostEqual(get_real(data, 4), -3.402823e12, delta=-3.402823e12 * -0.0000001)
+        self.assertAlmostEqual(get_real(data, 8), -175494351e-38, delta=-175494351e-38 * -0.0000001)
+        self.assertAlmostEqual(get_real(data, 12), -1.175494351e-12, delta=-1.175494351e-12 * -0.0000001)
+        self.assertAlmostEqual(get_real(data, 16), 0.0)
+        self.assertAlmostEqual(get_real(data, 20), 1.175494351e-38, delta=1.175494351e-38 * 0.0000001)
+        self.assertAlmostEqual(get_real(data, 24), 1.175494351e-12, delta=1.175494351e-12 * 0.0000001)
+        self.assertAlmostEqual(get_real(data, 28), 3.402823466e12, delta=3.402823466e12 * 0.0000001)
+        self.assertAlmostEqual(get_real(data, 32), 3.402823466e38, delta=3.402823466e38 * 0.0000001)
+
+    def test_read_string(self):
+        data = self.client.db_read(0, 100, 254)
+        self.assertEqual(get_string(data, 0, 254), "the brown fox jumps over the lazy dog")
+
+    def test_read_word(self):
+        data = self.client.db_read(0, 400, 4 * 4)
+        self.assertEqual(get_word(data, 0), 0x0000)
+        self.assertEqual(get_word(data, 4), 0x1234)
+        self.assertEqual(get_word(data, 8), 0xABCD)
+        self.assertEqual(get_word(data, 12), 0xFFFF)
+
+    def test_read_double_word(self):
+        data = self.client.db_read(0, 500, 8 * 4)
+        self.assertEqual(get_dword(data, 0), 0x00000000)
+        self.assertEqual(get_dword(data, 8), 0x12345678)
+        self.assertEqual(get_dword(data, 16), 0x1234ABCD)
+        self.assertEqual(get_dword(data, 24), 0xFFFFFFFF)
+
+
+if __name__ == '__main__':
+    import logging
+
+    logging.basicConfig()
+    unittest.main()