Update to v0.0.2

Added `packMessage` and `unpackMessage` for socket communications.
Tests are extended.

Author: iFA88

LICENSE.md

@@ -1,4 +1,4 @@
-FS message packer
+Fusion Solutions message packer
 Copyright (C) 2021 Fusion Solutions KFT <contact@fusionsolutions.io>
 
 This program is free software: you can redistribute it and/or modify

README.md

@@ -1,5 +1,5 @@
 [![Python package](https://github.com/FusionSolutions/python-fspacker/actions/workflows/python-package.yml/badge.svg)](https://github.com/FusionSolutions/python-fspacker/actions/workflows/python-package.yml)
-# FS Message packer
+# Fusion Solutions message packer
 
 ## Introduction
 

fsPacker/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.0.1"
+__version__ = "0.0.2"
 __doc__ = """
 FS Message Packer v{}
 Copyright (C) 2021 Fusion Solutions KFT <contact@fusionsolutions.io>
@@ -17,7 +17,7 @@
 along with this program.  If not, see <http://www.gnu.org/licenses/lgpl.txt>.
 """.format(__version__)
 from .fsPacker import (dump, dumps, load, loads, FSPackerError, UnsupportedType, UnsupportedVersion, InvalidOP,
-	MaxDictProtection, MaxOPProtection, OutOfData)
+	MaxDictProtection, MaxOPProtection, OutOfData, packMessage, unpackMessage)
 
 __all__ = ("dump", "dumps", "load", "loads", "FSPackerError", "UnsupportedType", "UnsupportedVersion", "InvalidOP",
-	"MaxDictProtection", "MaxOPProtection", "OutOfData")
+	"MaxDictProtection", "MaxOPProtection", "OutOfData", "packMessage", "unpackMessage")

fsPacker/__main__.py

@@ -20,6 +20,8 @@ def setUpClass(self) -> None:
 			0.0,
 			0.1,
 			-0.1,
+			1.234e+16,
+			1.234e-16,
 			"",
 			"test",
 			"Ő",
@@ -38,6 +40,7 @@ def test_dumpsAndLoads(self) -> None:
 		d:Any
 		for d in self.data:
 			self.assertEqual(loads(dumps( d )), d)
+		self.assertTupleEqual(loads(dumps( self.data )), self.data)
 		self.assertTupleEqual(loads(dumps( (self.data, self.data)) ), (self.data, self.data))
 		self.assertTupleEqual(loads(dumps( [self.data, self.data] )), (self.data, self.data))
 		self.assertDictEqual(loads(dumps( {"data":self.data} )), {"data":self.data})
@@ -60,5 +63,34 @@ def test_raises(self) -> None:
 		d = dumps(list(range(1024)))
 		with self.assertRaises(MaxDictProtection):
 			loads(d, maxDictSize=512)
-
+	def test_messagePackUnPack(self) -> None:
+		indicatorLength:int
+		messageLength:int
+		d:bytes = dumps(self.data)
+		packedMessage:bytes = packMessage(d)
+		indicatorLength, messageLength = unpackMessage(packedMessage)
+		self.assertEqual(indicatorLength, 1)
+		self.assertEqual(messageLength, len(d))
+		self.assertEqual(packedMessage[indicatorLength:], d)
+		#
+		packedMessage = packMessage(b"\x00"*0xFC)
+		indicatorLength, messageLength = unpackMessage(packedMessage)
+		self.assertEqual(indicatorLength, 1)
+		self.assertEqual(messageLength, 0xFC)
+		#
+		packedMessage = packMessage(b"\x00"*0xFD)
+		indicatorLength, messageLength = unpackMessage(packedMessage)
+		self.assertEqual(indicatorLength, 3)
+		self.assertEqual(messageLength, 0xFD)
+		#
+		packedMessage = packMessage(b"\x00"*0xFFFF)
+		indicatorLength, messageLength = unpackMessage(packedMessage)
+		self.assertEqual(indicatorLength, 3)
+		self.assertEqual(messageLength, 0xFFFF)
+		#
+		packedMessage = packMessage(b"\x00"*0xFFFFFF)
+		indicatorLength, messageLength = unpackMessage(packedMessage)
+		self.assertEqual(indicatorLength, 4)
+		self.assertEqual(messageLength, 0xFFFFFF)
+		#
 unittest.main(verbosity=2)

fsPacker/fsPacker.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 from io import BytesIO
 from math import ceil
-from typing import Any, Union, IO
+from typing import Dict, Any, List, Tuple, IO
 # Third party modules
 # Local modules
 # Program
@@ -41,6 +41,30 @@ class MaxOPProtection(FSPackerError): pass
 
 class OutOfData(FSPackerError): pass
 
+def packMessage(message:bytes) -> bytes:
+	d:int = len(message)
+	if d < 0xFD:
+		return d.to_bytes(1, VAR_ENDIANNESS) + message
+	elif d <= 0xFFFF:
+		return b"\xFD" + d.to_bytes(2, VAR_ENDIANNESS) + message
+	elif d <= 0xFFFFFF:
+		return b"\xFE" + d.to_bytes(3, VAR_ENDIANNESS) + message
+	elif d <= 0xFFFFFFFF:
+		return b"\xFF" + d.to_bytes(4, VAR_ENDIANNESS) + message
+	raise RuntimeError("Too big data to pack")
+
+def unpackMessage(buffer:bytes) -> Tuple[int, int]:
+	d:int = buffer[0]
+	if d < 0xFD:
+		return 1, d
+	elif d == 0xFD:
+		return 3, int.from_bytes(buffer[1:3], VAR_ENDIANNESS)
+	elif d == 0xFE:
+		return 4, int.from_bytes(buffer[1:4], VAR_ENDIANNESS)
+	elif d == 0xFF:
+		return 5, int.from_bytes(buffer[1:5], VAR_ENDIANNESS)
+	return 0, 0
+
 def _create_vint(d:int) -> bytes:
 	if d < 0xEC:
 		return d.to_bytes(1, VAR_ENDIANNESS)
@@ -52,18 +76,18 @@ def _create_vint(d:int) -> bytes:
 		return b"\xEE" + d.to_bytes(8, VAR_ENDIANNESS)
 
 class FSPacker:
+	def __init__(self) -> None:
+		self.dictCounter:int = 0
+		self.dictByKey:Dict[Any, int] = {}
+		self.dictBuffer:BytesIO = BytesIO()
+		self.opBuffer:BytesIO = BytesIO()
 	@classmethod
 	def dump(self, data:Any, fp:IO[bytes]) -> None:
 		fp.write( self()._parse(data) )
 		return
 	@classmethod
 	def dumps(self, data:Any) -> bytes:
 		return self()._parse(data)
-	def __init__(self):
-		self.dictCounter:int = 0
-		self.dictByKey:dict = {}
-		self.dictBuffer:BytesIO = BytesIO()
-		self.opBuffer:BytesIO = BytesIO()
 	def _parse(self, data:Any) -> bytes:
 		self._dump(data)
 		return VAR_VERSION + _create_vint(len(self.dictByKey)) + self.dictBuffer.getbuffer() + self.opBuffer.getbuffer()
@@ -149,15 +173,15 @@ def _register(self, k:Any) -> int:
 			self.dictCounter += 1
 		return self.dictByKey[k]
 
-class FSUnPacker:
+class FSUnpacker:
 	@classmethod
 	def load(self, data:IO[bytes], maxDictSize:int=0, maxOPSize:int=0) -> Any:
 		return self(BytesIO(data.read()), maxDictSize, maxOPSize)._parse()
 	@classmethod
 	def loads(self, data:bytes, maxDictSize:int=0, maxOPSize:int=0) -> Any:
 		return self(BytesIO(data), maxDictSize, maxOPSize)._parse()
 	def __init__(self, buffer:BytesIO, maxDictSize:int=0, maxOPSize:int=0):
-		self.dict:list = []
+		self.dict:List[Any] = []
 		self.buffer:BytesIO = buffer
 		self.maxDictSize:int = maxDictSize
 		self.maxOPSize:int = maxOPSize
@@ -246,5 +270,5 @@ def _loads(self) -> Any:
 
 dump = FSPacker.dump
 dumps = FSPacker.dumps
-load = FSUnPacker.load
-loads = FSUnPacker.loads
+load = FSUnpacker.load
+loads = FSUnpacker.loads

setup.py

@@ -11,7 +11,7 @@
 setup(
 	name = "python-fspacker",
 	version = fsPacker.__version__,
-	description = "FS message packer",
+	description = "Fusion Solutions message packer",
 	keywords = "message pack packer utility fusion solutions fusionsolutions",
 	author = "Andor `iFA88` Rajci - Fusions Solutions KFT",
 	author_email = "ifa@fusionsolutions.io",