add CAP LS/ACK/NEW/DEL support

ircstates/server.py

@@ -38,6 +38,10 @@ class Server(Named):
 
         self.isupport = ISupport()
 
+        self._temp_caps: Dict[str, Optional[str]]     = {}
+        self.caps: Optional[Dict[str, Optional[str]]] = None
+        self.agreed_caps: List[str]                   = []
+
     def __repr__(self) -> str:
         return f"Server(name={self.name!r})"
 
@@ -379,3 +383,40 @@ class Server(Named):
             user = self.users[nickname_lower]
             user.username = username
             user.hostname = hostname
+
+    @line_handler("CAP")
+    def handle_CAP(self, line: Line):
+        subcommand = line.params[1].upper()
+        multiline  = line.params[2] == "*"
+        caps       = line.params[2 + (1 if multiline else 0)]
+
+        tokens: Dict[str, Optional[str]] = {}
+        for cap in filter(bool, caps.split(" ")):
+            key, _, value = cap.partition("=")
+            tokens[key] = value or None
+
+        if subcommand == "LS":
+            self._temp_caps.update(tokens)
+            if not multiline:
+                self.caps = self._temp_caps.copy()
+                self._temp_caps.clear()
+        elif subcommand == "NEW":
+            if not self.caps == None:
+                self.caps.update(tokens)
+        elif subcommand == "DEL":
+            if not self.caps == None:
+                for key in tokens.keys():
+                    if key in self.caps:
+                        del self.caps[key]
+                        if key in self.agreed_caps:
+                            self.agreed_caps.remove(key)
+        elif subcommand == "ACK":
+            for key in tokens.keys():
+                if key.startswith("-"):
+                    key = key[1:]
+                    if key in self.agreed_caps:
+                        self.agreed_caps.remove(key)
+                elif (not key in self.agreed_caps and
+                        self.caps and
+                        key in self.caps):
+                    self.agreed_caps.append(key)

test/__init__.py

@@ -2,3 +2,4 @@ from .channel import *
 from .user    import *
 from .mode    import *
 from .motd    import *
+from .cap     import *

test/cap.py

@@ -0,0 +1,88 @@
+import unittest
+import ircstates, irctokens
+
+class CapTestLS(unittest.TestCase):
+    def test_one_line(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a b"))
+        self.assertEqual(server.caps, {"a": None, "b": None})
+
+    def test_two_lines(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS * :a b"))
+        self.assertEqual(server.caps, None)
+        server.parse_tokens(irctokens.tokenise("CAP * LS :c"))
+        self.assertEqual(server.caps, {"a": None, "b": None, "c": None})
+
+    def test_values(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a b= c=1"))
+        self.assertEqual(server.caps, {"a": None, "b": None, "c": "1"})
+
+class CapTestACK(unittest.TestCase):
+    def test_one_line(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a b"))
+        server.parse_tokens(irctokens.tokenise("CAP * ACK :a b"))
+        self.assertEqual(server.agreed_caps, ["a", "b"])
+
+    def test_two_lines(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a b c"))
+        server.parse_tokens(irctokens.tokenise("CAP * ACK * :a b"))
+        server.parse_tokens(irctokens.tokenise("CAP * ACK :c"))
+        self.assertEqual(server.agreed_caps, ["a", "b", "c"])
+
+    def test_not_ls(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS a"))
+        server.parse_tokens(irctokens.tokenise("CAP * ACK b"))
+        self.assertEqual(server.agreed_caps, [])
+
+class CapTestNEW(unittest.TestCase):
+    def test_no_ls(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * NEW :a"))
+        self.assertEqual(server.caps, None)
+
+    def test_one(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a"))
+        server.parse_tokens(irctokens.tokenise("CAP * NEW :b"))
+        self.assertEqual(server.caps, {"a": None, "b": None})
+
+    def test_two(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a"))
+        server.parse_tokens(irctokens.tokenise("CAP * NEW :b c"))
+        self.assertEqual(server.caps, {"a": None, "b": None, "c": None})
+
+class CapTestDEL(unittest.TestCase):
+    def test_not_acked(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * DEL a"))
+
+    def test_one_ls(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a"))
+        server.parse_tokens(irctokens.tokenise("CAP * ACK :a"))
+        server.parse_tokens(irctokens.tokenise("CAP * DEL :a"))
+        self.assertEqual(server.caps, {})
+        self.assertEqual(server.agreed_caps, [])
+
+    def test_two_ls(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a b"))
+        server.parse_tokens(irctokens.tokenise("CAP * ACK :a b"))
+        server.parse_tokens(irctokens.tokenise("CAP * DEL :a"))
+        self.assertEqual(server.caps, {"b": None})
+        self.assertEqual(server.agreed_caps, ["b"])
+
+    def test_two_del(self):
+        server = ircstates.Server("test")
+        server.parse_tokens(irctokens.tokenise("CAP * LS :a b"))
+        server.parse_tokens(irctokens.tokenise("CAP * ACK :a b"))
+        server.parse_tokens(irctokens.tokenise("CAP * DEL :a b"))
+        self.assertEqual(server.caps, {})
+        self.assertEqual(server.agreed_caps, [])
+