Split strategies for live and dry run.

subiquity/server/ad_joiner.py

@@ -20,10 +20,43 @@ from subiquity.common.types import (
 )
 
 
+class AdJoinStrategy():
+    cmd = "/usr/sbin/realm"
+    args = ["join"]
+
+    async def do_join(self, info: ADConnectionInfo) -> AdJoinResult:
+        # Now what?
+        # TODO: Join.
+        result = AdJoinResult.JOIN_ERROR
+        return await asyncio.sleep(3, result=result)
+
+
+class StubStrategy(AdJoinStrategy):
+    async def do_join(self, info: ADConnectionInfo, hostname: str, context) \
+            -> AdJoinResult:
+        """ Enables testing without real join. The result depends on the
+            domain name initial character, such that if it is:
+            - p or P: returns PAM_ERROR.
+            - j or J: returns JOIN_ERROR.
+            - returns OK otherwise. """
+        initial = info.domain_name[0]
+        if initial in ('j', 'J'):
+            return AdJoinResult.JOIN_ERROR
+
+        if initial in ('p', 'P'):
+            return AdJoinResult.PAM_ERROR
+
+        return AdJoinResult.OK
+
+
 class AdJoiner():
-    def __init__(self):
+    def __init__(self, dry_run: bool):
         self._result = AdJoinResult.UNKNOWN
         self.join_task = None
+        if dry_run:
+            self.strategy = StubStrategy()
+        else:
+            self.strategy = AdJoinStrategy()
 
     async def join_domain(self, info: ADConnectionInfo) -> AdJoinResult:
         self.join_task = asyncio.create_task(self.async_join(info))
@@ -31,8 +64,7 @@ class AdJoiner():
         return self._result
 
     async def async_join(self, info: ADConnectionInfo) -> AdJoinResult:
-        # TODO: Join.
-        return await asyncio.sleep(3, result=AdJoinResult.JOIN_ERROR)
+        return await self.strategy.do_join(info)
 
     async def join_result(self):
         if self.join_task is None:

subiquity/server/controllers/ad.py

@@ -158,7 +158,7 @@ class ADController(SubiquityController):
     async def join_domain(self) -> None:
         """To be called from the install controller if joining was requested"""
         if self.ad_joiner is None:
-            self.ad_joiner = AdJoiner()
+            self.ad_joiner = AdJoiner(self.app.opts.dry_run)
 
         await self.ad_joiner.join_domain(self.model.conn_info)
 

subiquity/tests/api/test_api.py

@@ -1697,5 +1697,12 @@ class TestActiveDirectory(TestAPI):
                                          data='$Ubuntu')
             self.assertEqual('OK', result)
             # Attempts to join with the info supplied above.
+            ad_dict = {
+                'admin_name': 'Ubuntu',
+                'domain_name': 'jubuntu.com',
+                'password': 'u',
+            }
+            result = await instance.post(endpoint, ad_dict)
+            self.assertIsNone(result)
             join_result = await instance.get(endpoint + '/join_result')
             self.assertEqual('JOIN_ERROR', join_result)