Kivy Mock backend added

This commit is contained in:
shekhar-cis 2022-01-06 17:32:07 +05:30
parent ea16d6fefa
commit 4ec5eaf016
Signed by untrusted user: shekhar-cis
GPG Key ID: F4F00AB04E83F9A7
15 changed files with 153 additions and 0 deletions

View File

View File

@ -0,0 +1,32 @@
# pylint: disable=no-name-in-module, import-error
"""
Bitmessage mock
"""
from pybitmessage.class_addressGenerator import addressGenerator
from pybitmessage.inventory import Inventory
from pybitmessage.mpybit import NavigateApp
from pybitmessage import state
class MockMain(object): # pylint: disable=too-few-public-methods
"""Mock main function"""
def __init__(self):
"""Start main application"""
addressGeneratorThread = addressGenerator()
# close the main program even if there are threads left
addressGeneratorThread.start()
Inventory()
state.kivyapp = NavigateApp()
state.kivyapp.run()
def main():
"""Triggers main module"""
MockMain()
if __name__ == "__main__":
main()

1
src/tests/mock/images Symbolic link
View File

@ -0,0 +1 @@
../../images/

View File

@ -0,0 +1,8 @@
"""Mock kivy app with mock threads."""
from pybitmessage import state
if __name__ == '__main__':
state.kivy = True
print("Kivy Loading......")
from bitmessagemock import main
main()

View File

View File

@ -0,0 +1 @@
../../../addresses.py

View File

@ -0,0 +1 @@
../../../bmconfigparser.py

View File

@ -0,0 +1,80 @@
"""
A thread for creating addresses
"""
from six.moves import queue
from pybitmessage import state
from pybitmessage import queues
from pybitmessage.bmconfigparser import BMConfigParser
from pybitmessage.network.threads import StoppableThread
fake_addresses = {
'BM-2cUgQGcTLWAkC6dNsv2Bc8XB3Y1GEesVLV': {
'privsigningkey': '5KWXwYq1oJMzghUSJaJoWPn8VdeBbhDN8zFot1cBd6ezKKReqBd',
'privencryptionkey': '5JaeFJs8iPcQT3N8676r3gHKvJ5mTWXy1VLhGCEDqRs4vpvpxV8'
},
'BM-2cUd2dm8MVMokruMTcGhhteTpyRZCAMhnA': {
'privsigningkey': '5JnJ79nkcwjo4Aj7iG8sFMkzYoQqWfpUjTcitTuFJZ1YKHZz98J',
'privencryptionkey': '5JXgNzTRouFLqSRFJvuHMDHCYPBvTeMPBiHt4Jeb6smNjhUNTYq'
},
'BM-2cWyvL54WytfALrJHZqbsDHca5QkrtByAW': {
'privsigningkey': '5KVE4gLmcfYVicLdgyD4GmnbBTFSnY7Yj2UCuytQqgBBsfwDhpi',
'privencryptionkey': '5JTw48CGm5CP8fyJUJQMq8HQANQMHDHp2ETUe1dgm6EFpT1egD7'
},
'BM-2cTE65PK9Y4AQEkCZbazV86pcQACocnRXd': {
'privsigningkey': '5KCuyReHx9MB4m5hhEyCWcLEXqc8rxhD1T2VWk8CicPFc8B6LaZ',
'privencryptionkey': '5KBRpwXdX3n2tP7f583SbFgfzgs6Jemx7qfYqhdH7B1Vhe2jqY6'
},
'BM-2cX5z1EgmJ87f2oKAwXdv4VQtEVwr2V3BG': {
'privsigningkey': '5K5UK7qED7F1uWCVsehudQrszLyMZxFVnP6vN2VDQAjtn5qnyRK',
'privencryptionkey': '5J5coocoJBX6hy5DFTWKtyEgPmADpSwfQTazMpU7QPeART6oMAu'
}
}
class addressGenerator(StoppableThread):
"""A thread for creating fake addresses"""
name = "addressGenerator"
address_list = list(fake_addresses.keys())
def stopThread(self):
""""To stop address generator thread"""
try:
queues.addressGeneratorQueue.put(("stopThread", "data"))
except queue.Full:
self.logger.warning('addressGeneratorQueue is Full')
super(addressGenerator, self).stopThread() # pylint: disable=super-with-arguments
def run(self):
"""
Process the requests for addresses generation
from `.queues.addressGeneratorQueue`
"""
while state.shutdown == 0:
queueValue = queues.addressGeneratorQueue.get()
try:
address = self.address_list.pop(0)
except IndexError:
self.logger.error(
'Program error: you can only create 5 fake addresses')
continue
if len(queueValue) >= 3:
label = queueValue[3]
else:
label = ''
BMConfigParser().add_section(address)
BMConfigParser().set(address, 'label', label)
BMConfigParser().set(address, 'enabled', 'true')
BMConfigParser().set(
address, 'privsigningkey', fake_addresses[address]['privsigningkey'])
BMConfigParser().set(
address, 'privencryptionkey', fake_addresses[address]['privencryptionkey'])
BMConfigParser().save()
queues.addressGeneratorQueue.task_done()

View File

@ -0,0 +1,15 @@
"""The Inventory singleton"""
# TODO make this dynamic, and watch out for frozen, like with messagetypes
from pybitmessage.singleton import Singleton
# pylint: disable=old-style-class,too-few-public-methods
@Singleton
class Inventory():
"""
Inventory singleton class which uses storage backends
to manage the inventory.
"""
def __init__(self):
self.numberOfInventoryLookupsPerformed = 0

View File

@ -0,0 +1 @@
../../../../network/threads.py

View File

@ -0,0 +1 @@
../../../queues.py

View File

@ -0,0 +1,11 @@
# pylint: disable=invalid-name
"""shutdown function"""
from pybitmessage import state
def doCleanShutdown():
"""
Used to exit Kivy UI.
"""
state.shutdown = 1

View File

@ -0,0 +1 @@
../../../singleton.py

View File

@ -0,0 +1 @@
../../../state.py