added test for learning lan services

src/lan/__init__.py

@@ -31,8 +31,11 @@ class LANManager:
 
     def __init__(self, too_many: "TooMany"):
         self.too_many = too_many
+        self.peers: "exploded IP Address string" = set()
+
 
     def start(self):
         Thread(target=learn_services, daemon=True).start()
         Thread(target=advertise_service, daemon=True).start()
+        Thread(tra)
 

src/lan/discover.py

@@ -4,6 +4,10 @@ Discover and publish private-network
 """
 import socket
 import struct
+from typing import TYPE_CHECKING
+from typing import List
+from ipaddress import ip_address
+from socket import SHUT_RDWR
 
 from .getip import lan_ips
 from utils.bettersleep import better_sleep
@@ -27,7 +31,10 @@ IS_ALL_GROUPS = True
 ANNOUNCE_LOOP_SLEEP = 30
 
 
-def learn_services():
+
+def learn_services(lan_service_list: List):
+    """Take a list to infintely add lan service info to."""
+
     sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     if IS_ALL_GROUPS:
@@ -41,11 +48,26 @@ def learn_services():
     sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
 
     while True:
-        rec_data = sock.recv(1024)
+        service_ips = sock.recv(200).decode('utf-8')
-        print('reced_data', rec_data)
+        if 'onionr' not in service_ips:
+            continue
+        service_ips = service_ips.replace('onionr-', '').split('-')
+        port = 0
+        for service in service_ips:
+            try:
+                ip_address(service)
+            except ValueError:
+                service_ips.remove(service)
+                continue
+        # remove our own ips
+        service_ips = set(lan_ips) ^ set(service_ips)
+        # remove known ips and add to external list
+        lan_service_list = set(service_ips) ^ set(lan_service_list)
 
-    sock.shutdown()
+
+    sock.shutdown(SHUT_RDWR)
     sock.close()
+    # no return intended, list modified by reference
 
 
 def advertise_service():
@@ -61,5 +83,5 @@ def advertise_service():
     while True:
         sock.sendto(f"onionr-{ips}".encode('utf-8'), (MCAST_GRP, MCAST_PORT))
         better_sleep(ANNOUNCE_LOOP_SLEEP)
-    sock.shutdown()
+    sock.shutdown(SHUT_RDWR)
     sock.close()

src/lan/server/__init__.py

@@ -2,6 +2,8 @@
 
 LAN transport server thread
 """
+from gevent.pywsgi import WSGIServer
+from flask import Flask
 """
     This program is free software: you can redistribute it and/or modify
     it under the terms of the GNU General Public License as published by
@@ -16,3 +18,16 @@ LAN transport server thread
     You should have received a copy of the GNU General Public License
     along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
+
+
+class LANServer:
+    def __init__(self, shared_state):
+        app = Flask(__name__)
+        self.app = app
+
+        @app.route("/")
+        def ping():
+            return "pong!"
+
+    def start_server():
+        return

tests/integration-tests/test_lan_discover.py

@@ -1,35 +0,0 @@
-from unittest.mock import patch
-import sys, os
-sys.path.append(".")
-sys.path.append("src/")
-import unittest, uuid
-TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
-print("Test directory:", TEST_DIR)
-os.environ["ONIONR_HOME"] = TEST_DIR
-from utils import createdirs
-from onionrcommands import parser
-import onionrsetup as setup
-from netcontroller.torcontrol import customtorrc
-from utils import createdirs
-from onionrsetup import setup_config, setup_default_plugins
-from coredb import blockmetadb
-from etc.onionrvalues import BLOCK_EXPORT_FILE_EXT
-from threading import Thread
-
-createdirs.create_dirs()
-setup_config()
-setup_default_plugins()
-import config
-from filepaths import export_location
-
-class OnionrTests(unittest.TestCase):
-    def test_lan_discover(self):
-        testargs = ['onionr.py', 'start']
-        with patch.object(sys, 'argv', testargs):
-            try:
-                Thread(target=parser.register, daemon=True).start()
-            except SystemExit:
-                pass
-
-
-unittest.main()

tests/test_lan_learn.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+import sys, os
+sys.path.append(".")
+sys.path.append("src/")
+import uuid
+from threading import Thread
+TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
+print("Test directory:", TEST_DIR)
+os.environ["ONIONR_HOME"] = TEST_DIR
+import unittest, json
+
+from utils import identifyhome, createdirs
+from onionrsetup import setup_config
+createdirs.create_dirs()
+setup_config()
+from utils import bettersleep
+from lan.discover import lan_ips, MCAST_GRP, MCAST_PORT
+from lan.discover import learn_services, advertise_service
+import socket
+from socket import SHUT_RDWR
+lan_ips = ['']
+
+
+class TestLanLearn(unittest.TestCase):
+    def test_lan_learn(self):
+        test_ip = '192.168.1.30'
+        def multicast():
+            port = 1349
+            MULTICAST_TTL = 3
+            ips = '-'.join([test_ip]) + f'-{port}'
+
+            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
+            sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, MULTICAST_TTL)
+            sock.sendto(f"onionr-{ips}".encode('utf-8'), (MCAST_GRP, MCAST_PORT))
+            bettersleep.better_sleep(1)
+            try:
+                sock.shutdown(SHUT_RDWR)
+            except OSError:
+                pass
+            sock.close()
+        test_list = [test_ip]
+
+        Thread(target=learn_services, args=[test_list], daemon=True).start()
+        bettersleep.better_sleep(3)
+        multicast()
+        self.assertIn(test_ip, test_list)
+
+unittest.main()