Implement create_appservice_txn with tests.

synapse/storage/appservice.py

@@ -17,9 +17,10 @@ from simplejson import JSONDecodeError
 import simplejson as json
 from twisted.internet import defer
 
+from syutil.jsonutil import encode_canonical_json
 from synapse.api.constants import Membership
 from synapse.api.errors import StoreError
-from synapse.appservice import ApplicationService, ApplicationServiceState
+from synapse.appservice import ApplicationService, AppServiceTransaction
 from synapse.storage.roommember import RoomsForUser
 from ._base import SQLBaseStore
 
@@ -417,9 +418,46 @@ class ApplicationServiceTransactionStore(SQLBaseStore):
         Returns:
             AppServiceTransaction: A new transaction.
         """
-        # TODO: work out txn id (highest txn id for this service += 1)
-        # TODO: Within same db transaction, Insert new txn into txn table
-        pass
+        return self.runInteraction(
+            "create_appservice_txn",
+            self._create_appservice_txn,
+            service, events
+        )
+
+    def _create_appservice_txn(self, txn, service, events):
+        # work out new txn id (highest txn id for this service += 1)
+        # The highest id may be the last one sent (in which case it is last_txn)
+        # or it may be the highest in the txns list (which are waiting to be/are
+        # being sent)
+        result = txn.execute(
+            "SELECT last_txn FROM application_services_state WHERE as_id=?",
+            (service.id,)
+        )
+        last_txn_id = result.fetchone()
+        if last_txn_id is None:  # no row exists
+            last_txn_id = 0
+        else:
+            last_txn_id = int(last_txn_id[0])  # select 'last_txn' col
+
+        result = txn.execute(
+            "SELECT MAX(txn_id) FROM application_services_txns WHERE as_id=?",
+            (service.id,)
+        )
+        highest_txn_id = result.fetchone()[0]
+        if highest_txn_id is None:
+            highest_txn_id = 0
+
+        new_txn_id = max(highest_txn_id, last_txn_id) + 1
+
+        # Insert new txn into txn table
+        txn.execute(
+            "INSERT INTO application_services_txns(as_id, txn_id, content) "
+            "VALUES(?,?,?)",
+            (service.id, new_txn_id, encode_canonical_json(events))
+        )
+        return AppServiceTransaction(
+            service=service, id=new_txn_id, events=events
+        )
 
     def complete_appservice_txn(self, txn_id, service):
         """Completes an application service transaction.

tests/storage/test_appservice.py

@@ -21,6 +21,7 @@ from synapse.storage.appservice import (
     ApplicationServiceStore, ApplicationServiceTransactionStore
 )
 
+import json
 from mock import Mock
 from tests.utils import SQLiteMemoryDbPool, MockClock
 
@@ -166,6 +167,20 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase):
             (id, state, txn)
         )
 
+    def _insert_txn(self, as_id, txn_id, content):
+        return self.db_pool.runQuery(
+            "INSERT INTO application_services_txns(as_id, txn_id, content) "
+            "VALUES(?,?,?)",
+            (as_id, txn_id, json.dumps(content))
+        )
+
+    def _set_last_txn(self, as_id, txn_id):
+        return self.db_pool.runQuery(
+            "INSERT INTO application_services_state(as_id, last_txn, state) "
+            "VALUES(?,?,?)",
+            (as_id, txn_id, ApplicationServiceState.UP)
+        )
+
     @defer.inlineCallbacks
     def test_get_appservice_state_none(self):
         service = Mock(id=999)
@@ -237,6 +252,58 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase):
         )
         self.assertEquals(service.id, rows[0][0])
 
+    @defer.inlineCallbacks
+    def test_create_appservice_txn_first(self):
+        service = Mock(id=self.as_list[0]["id"])
+        events = [{"type": "nothing"}, {"type": "here"}]
+        txn = yield self.store.create_appservice_txn(service, events)
+        self.assertEquals(txn.id, 1)
+        self.assertEquals(txn.events, events)
+        self.assertEquals(txn.service, service)
+
+    @defer.inlineCallbacks
+    def test_create_appservice_txn_older_last_txn(self):
+        service = Mock(id=self.as_list[0]["id"])
+        events = [{"type": "nothing"}, {"type": "here"}]
+        yield self._set_last_txn(service.id, 9643)  # AS is falling behind
+        yield self._insert_txn(service.id, 9644, events)
+        yield self._insert_txn(service.id, 9645, events)
+        txn = yield self.store.create_appservice_txn(service, events)
+        self.assertEquals(txn.id, 9646)
+        self.assertEquals(txn.events, events)
+        self.assertEquals(txn.service, service)
+
+    @defer.inlineCallbacks
+    def test_create_appservice_txn_up_to_date_last_txn(self):
+        service = Mock(id=self.as_list[0]["id"])
+        events = [{"type": "nothing"}, {"type": "here"}]
+        yield self._set_last_txn(service.id, 9643)
+        txn = yield self.store.create_appservice_txn(service, events)
+        self.assertEquals(txn.id, 9644)
+        self.assertEquals(txn.events, events)
+        self.assertEquals(txn.service, service)
+
+    @defer.inlineCallbacks
+    def test_create_appservice_txn_up_fuzzing(self):
+        service = Mock(id=self.as_list[0]["id"])
+        events = [{"type": "nothing"}, {"type": "here"}]
+        yield self._set_last_txn(service.id, 9643)
+
+        # dump in rows with higher IDs to make sure the queries aren't wrong.
+        yield self._set_last_txn(self.as_list[1]["id"], 119643)
+        yield self._set_last_txn(self.as_list[2]["id"], 9)
+        yield self._set_last_txn(self.as_list[3]["id"], 9643)
+        yield self._insert_txn(self.as_list[1]["id"], 119644, events)
+        yield self._insert_txn(self.as_list[1]["id"], 119645, events)
+        yield self._insert_txn(self.as_list[1]["id"], 119646, events)
+        yield self._insert_txn(self.as_list[2]["id"], 10, events)
+        yield self._insert_txn(self.as_list[3]["id"], 9643, events)
+
+        txn = yield self.store.create_appservice_txn(service, events)
+        self.assertEquals(txn.id, 9644)
+        self.assertEquals(txn.events, events)
+        self.assertEquals(txn.service, service)
+
     @defer.inlineCallbacks
     def test_get_appservices_by_state_single(self):
         yield self._set_state(