WIP type defintions

mypy.ini

+[mypy]
+ignore_missing_imports = True
+
+[mypy-slapos]
+ignore_missing_imports = True
+[mypy-slapos.slap]
+ignore_missing_imports = True
+[mypy-slapos.slap.grid]
+ignore_missing_imports = True
+
+
+# interfaces cause error: Method must have at least one argument
+[mypy-slapos.slap.interface]
+ignore_errors = True
+[mypy-slapos.slap.interface.slap]
+ignore_errors = True
+

setup.py

@@ -75,7 +75,8 @@ setup(name=name,
           'cachecontrol',
           'lockfile',
           'uritemplate', # used by hateoas navigator
-          'subprocess32; python_version<"3"'
+          'subprocess32; python_version<"3"',
+          'typing; python_version<"3"',
         ] + additional_install_requires,
       extras_require=extras_require,
       tests_require=extras_require['test'],

slapos/__init__.py

@@ -3,4 +3,4 @@ try:
     __import__('pkg_resources').declare_namespace(__name__)
 except ImportError:
     from pkgutil import extend_path
-    __path__ = extend_path(__path__, __name__)
+    __path__ = extend_path(__path__, __name__) # type: ignore

slapos/cli/proxy_show.py

@@ -44,7 +44,7 @@ from slapos.proxy import ProxyConfig
 from slapos.proxy.db_version import DB_VERSION
 from slapos.util import sqlite_connect, str2bytes
 
-if bytes is str:
+if sys.version_info[0] == 3:
   from io import BytesIO
   class StringIO(BytesIO):
     # Something between strict io.BytesIO and laxist/slow StringIO.StringIO

slapos/collect/temperature/__init__.py

 from __future__ import print_function
 
 from multiprocessing import Process, active_children, cpu_count, Pipe
-try:
-  import subprocess32 as subprocess
-except ImportError:
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
   import subprocess
+else:
+  try:
+    import subprocess32 as subprocess
+  except ImportError:
+    import subprocess
 import os
 import signal
 import sys

slapos/grid/SlapObject.py

@@ -38,7 +38,8 @@ import subprocess
 import tarfile
 import tempfile
 import time
-from six.moves import xmlrpc_client as xmlrpclib, range
+from six.moves import xmlrpc_client as xmlrpclib # type: ignore
+from six.moves import range
 from six.moves.configparser import ConfigParser
 
 from supervisor import xmlrpc

slapos/grid/slapgrid.py

@@ -49,6 +49,10 @@ if sys.version_info < (2, 6):
   warnings.warn('Used python version (%s) is old and has problems with'
       ' IPv6 connections' % sys.version.split('\n')[0])
 
+from typing import List, Tuple, Sequence, TYPE_CHECKING
+if TYPE_CHECKING:
+  from ..slap.slap import ComputerPartition
+
 from lxml import etree
 
 from slapos import manager as slapmanager
@@ -537,6 +541,7 @@ stderr_logfile_backups=1
       launchSupervisord(instance_root=self.instance_root, logger=self.logger)
 
   def getComputerPartitionList(self):
+    # type: () -> Sequence[ComputerPartition]
     try:
       return self.computer.getComputerPartitionList()
     except socket.error as exc:
@@ -789,7 +794,7 @@ stderr_logfile_backups=1
       reload_process = FPopen(reload_cmd, universal_newlines=True)
       stdout, stderr = reload_process.communicate()
       if reload_process.returncode != 0:
-        raise Exception("Failed to load firewalld rules with command %s.\n%" % (
+        raise Exception("Failed to load firewalld rules with command %s.\n%s" % (
                         stderr, reload_cmd))
 
       with open(firewall_rules_path, 'w') as frules:
@@ -1254,6 +1259,7 @@ stderr_logfile_backups=1
         f.write(str(timestamp))
 
   def FilterComputerPartitionList(self, computer_partition_list):
+    # type: (Sequence[ComputerPartition]) -> List[ComputerPartition]
     """
     Try to filter valid partitions to be processed from free partitions.
     """
@@ -1392,12 +1398,12 @@ stderr_logfile_backups=1
     self.logger.info('=' * 80)
     if process_error_partition_list:
       self.logger.info('Error while processing the following partitions:')
-      for partition, exc in process_error_partition_list:
-        self.logger.info('  %s[%s]: %s', partition.getId(), getPartitionType(partition), exc)
+      for partition, error in process_error_partition_list:
+        self.logger.info('  %s[%s]: %s', partition.getId(), getPartitionType(partition), error)
     if promise_error_partition_list:
       self.logger.info('Error with promises for the following partitions:')
-      for partition, exc in promise_error_partition_list:
-        self.logger.info('  %s[%s]: %s', partition.getId(), getPartitionType(partition), exc)
+      for partition, error in promise_error_partition_list:
+        self.logger.info('  %s[%s]: %s', partition.getId(), getPartitionType(partition), error)
 
     # Return success value
     if not clean_run:
@@ -1420,7 +1426,7 @@ stderr_logfile_backups=1
     computer_partition_list = self.FilterComputerPartitionList(
         self.getComputerPartitionList())
 
-    promise_error_partition_list = []
+    promise_error_partition_list = [] # type: List[Tuple[ComputerPartition, Union[PromiseError, Exception]]]
     for computer_partition in computer_partition_list:
       try:
         # Process the partition itself
@@ -1444,8 +1450,8 @@ stderr_logfile_backups=1
 
     if promise_error_partition_list:
       self.logger.info('Finished computer partitions.')
-      for partition, exc in promise_error_partition_list:
-        self.logger.info('  %s[%s]: %s', partition.getId(), getPartitionType(partition), exc)
+      for partition, error in promise_error_partition_list:
+        self.logger.info('  %s[%s]: %s', partition.getId(), getPartitionType(partition), error)
 
     # Return success value
     if not clean_run_promise:

slapos/grid/svcbackend.py

@@ -35,7 +35,7 @@ import subprocess
 import stat
 import sys
 import time
-from six.moves import xmlrpc_client as xmlrpclib
+from six.moves import xmlrpc_client as xmlrpclib # type: ignore
 import contextlib
 
 from slapos.grid.utils import (createPrivateDirectory, SlapPopen, updateFile)

slapos/proxy/views.py

@@ -53,8 +53,11 @@ EMPTY_DICT_XML = dumps({})
 class UnauthorizedError(Exception):
   pass
 
+from typing import Dict, Union, no_type_check
 
+@no_type_check
 def partitiondict2partition(partition):
+  # type: (Dict[str, str]) -> ComputerPartition
   slap_partition = ComputerPartition(partition['computer_reference'],
       partition['reference'])
   slap_partition._software_release_document = None
@@ -365,6 +368,7 @@ def supplySupply():
 
 @app.route('/requestComputerPartition', methods=['POST'])
 def requestComputerPartition():
+  # type: () -> str
   parsed_request_dict = parseRequestComputerPartitionForm(request.form)
   # Is it a slave instance?
   slave = loads(request.form.get('shared_xml', EMPTY_DICT_XML).encode('utf-8'))
@@ -457,6 +461,7 @@ def requestComputerPartition():
   return dumps(software_instance)
 
 def parseRequestComputerPartitionForm(form):
+  # type: (Dict) -> Dict
   """
   Parse without intelligence a form from a request(), return it.
   """
@@ -469,7 +474,7 @@ def parseRequestComputerPartitionForm(form):
     'filter_kw': loads(form.get('filter_xml', EMPTY_DICT_XML).encode('utf-8')),
     # Note: currently ignored for slave instance (slave instances
     # are always started).
-    'requested_state': loads(form.get('state').encode('utf-8')),
+    'requested_state': loads(form['state'].encode('utf-8')),
   }
 
   return parsed_dict
@@ -543,10 +548,11 @@ def isRequestToBeForwardedToExternalMaster(parsed_request_dict):
     return None
 
 def forwardRequestToExternalMaster(master_url, request_form):
+  # type: (str, Dict) -> str
   """
   Forward instance request to external SlapOS Master.
   """
-  master_entry = app.config.get('multimaster').get(master_url, {})
+  master_entry = app.config['multimaster'].get(master_url, {})
   key_file = master_entry.get('key')
   cert_file = master_entry.get('cert')
   if master_url.startswith('https') and (not key_file or not cert_file):
@@ -568,7 +574,7 @@ def forwardRequestToExternalMaster(master_url, request_form):
              {'partition_reference':partition_reference, 'master_url': master_url})
 
   new_request_form = request_form.copy()
-  filter_kw = loads(new_request_form['filter_xml'].encode('utf-8'))
+  filter_kw = loads(request_form['filter_xml'].encode('utf-8'))
   filter_kw['source_instance_id'] = partition_reference
   new_request_form['filter_xml'] = dumps(filter_kw)
 
@@ -576,7 +582,7 @@ def forwardRequestToExternalMaster(master_url, request_form):
   partition = loads(xml)
 
   # XXX move to other end
-  partition._master_url = master_url
+  partition._master_url = master_url # type: ignore
 
   return dumps(partition)
 

slapos/slap/slap.py

@@ -42,6 +42,7 @@ import re
 from functools import wraps
 
 import six
+from typing import Any, Dict, Tuple, List, Optional, Union
 
 from .exception import ResourceNotReady, ServerError, NotFoundError, \
           ConnectionError
@@ -67,6 +68,17 @@ fallback_handler = logging.StreamHandler()
 fallback_logger.setLevel(logging.INFO)
 fallback_logger.addHandler(fallback_handler)
 
+SoftwareState = str
+InstanceState = str
+
+PartitionParameters = Dict[str, str]
+FilterParameters = Dict[str, str]
+
+# XXX better types if typing supports it ?
+import typing
+if typing.TYPE_CHECKING and hasattr(typing, 'Literal'):
+  SoftwareState = typing.Literal["available", "destroyed"] # type: ignore
+  InstanceState = typing.Literal["available", "destroyed"] # type: ignore
 
 DEFAULT_SOFTWARE_TYPE = 'RootSoftwareInstance'
 COMPUTER_PARTITION_REQUEST_LIST_TEMPLATE_FILENAME = '.slapos-request-transaction-%s'
@@ -133,12 +145,14 @@ class SoftwareRelease(SlapDocument):
     return (self._software_release, self._computer_guid, )
 
   def getComputerId(self):
+    # type: () -> str
     if not self._computer_guid:
       raise NameError('computer_guid has not been defined.')
     else:
       return self._computer_guid
 
   def getURI(self):
+    # type: () -> str
     if not self._software_release:
       raise NameError('software_release has not been defined.')
     else:
@@ -155,6 +169,7 @@ class SoftwareRelease(SlapDocument):
       (logger or fallback_logger).exception('')
 
   def available(self):
+    # type: () -> None
     if getattr(self, '_known_state', 'unknown') != "available":
       # Not required to repost if not needed.
       self._connection_helper.POST('availableSoftwareRelease', data={
@@ -162,17 +177,20 @@ class SoftwareRelease(SlapDocument):
         'computer_id': self.getComputerId()})
 
   def building(self):
+    # type: () -> None
     if getattr(self, '_known_state', 'unknown') != "building":
       self._connection_helper.POST('buildingSoftwareRelease', data={
         'url': self.getURI(),
         'computer_id': self.getComputerId()})
 
   def destroyed(self):
+    # type: () -> None
     self._connection_helper.POST('destroyedSoftwareRelease', data={
       'url': self.getURI(),
       'computer_id': self.getComputerId()})
 
   def getState(self):
+    # type: () -> SoftwareState
     return getattr(self, '_requested_state', 'available')
 
 
@@ -217,6 +235,7 @@ class SoftwareInstance(SlapDocument):
 class Supply(SlapDocument):
 
   def supply(self, software_release, computer_guid=None, state='available'):
+    # type: (str, Optional[str], SoftwareState) -> None
     try:
       self._connection_helper.POST('supplySupply', data={
         'url': software_release,
@@ -229,6 +248,7 @@ class Supply(SlapDocument):
 @implementer(interface.IToken)
 class Token(SlapDocument):
   def request(self):
+    # type: () -> Token
     return self._hateoas_navigator.getToken()
 
 @implementer(interface.IOpenOrder)
@@ -237,7 +257,7 @@ class OpenOrder(SlapRequester):
   def request(self, software_release, partition_reference,
               partition_parameter_kw=None, software_type=None,
               filter_kw=None, state=None, shared=False):
-
+    # type: (str, str, Optional[PartitionParameters], Optional[str], Optional[FilterParameters], Optional[InstanceState], bool) -> ComputerPartition
     if partition_parameter_kw is None:
       partition_parameter_kw = {}
     elif not isinstance(partition_parameter_kw, dict):
@@ -321,12 +341,15 @@ class Computer(SlapDocument):
   def __init__(self, computer_id, connection_helper=None, hateoas_navigator=None):
     SlapDocument.__init__(self, connection_helper, hateoas_navigator)
     self._computer_id = computer_id
+    self._software_release_list = None # type: List[SoftwareRelease]
+    self._computer_partition_list = None # type: List[ComputerPartition]
 
   def __getinitargs__(self):
     return (self._computer_id, )
 
   @_syncComputerInformation
   def getSoftwareReleaseList(self):
+    # type: () -> List[SoftwareRelease]
     """
     Returns the list of software release which has to be supplied by the
     computer.
@@ -340,6 +363,7 @@ class Computer(SlapDocument):
 
   @_syncComputerInformation
   def getComputerPartitionList(self):
+    # type: () -> List[ComputerPartition]
     for computer_partition in self._computer_partition_list:
       computer_partition._connection_helper = self._connection_helper
       computer_partition._hateoas_navigator = self._hateoas_navigator
@@ -356,6 +380,7 @@ class Computer(SlapDocument):
     return self._connection_helper.POST('loadComputerConfigurationFromXML', data={'xml': xml})
 
   def bang(self, message):
+    # type: (str) -> None
     self._connection_helper.POST('computerBang', data={
       'computer_id': self._computer_id,
       'message': message})
@@ -365,10 +390,12 @@ class Computer(SlapDocument):
     return loads(xml)
 
   def revokeCertificate(self):
+    # type: () -> None
     self._connection_helper.POST('revokeComputerCertificate', data={
       'computer_id': self._computer_id})
 
   def generateCertificate(self):
+    # type: () -> str
     xml = self._connection_helper.POST('generateComputerCertificate', data={
       'computer_id': self._computer_id})
     return loads(xml)
@@ -451,6 +478,7 @@ class ComputerPartition(SlapRequester):
   def request(self, software_release, software_type, partition_reference,
               shared=False, partition_parameter_kw=None, filter_kw=None,
               state=None):
+    # type: (str, str, str, bool, Optional[PartitionParameters], Optional[FilterParameters], Optional[InstanceState]) -> ComputerPartition
     if partition_parameter_kw is None:
       partition_parameter_kw = {}
     elif not isinstance(partition_parameter_kw, dict):
@@ -482,18 +510,21 @@ class ComputerPartition(SlapRequester):
     return self._requestComputerPartition(request_dict)
 
   def destroyed(self):
+    # type: () -> None
     self._connection_helper.POST('destroyedComputerPartition', data={
       'computer_id': self._computer_id,
       'computer_partition_id': self.getId(),
       })
 
   def started(self):
+    # type: () -> None
     self._connection_helper.POST('startedComputerPartition', data={
       'computer_id': self._computer_id,
       'computer_partition_id': self.getId(),
       })
 
   def stopped(self):
+    # type: () -> None
     self._connection_helper.POST('stoppedComputerPartition', data={
       'computer_id': self._computer_id,
       'computer_partition_id': self.getId(),
@@ -509,6 +540,7 @@ class ComputerPartition(SlapRequester):
       (logger or fallback_logger).exception('')
 
   def bang(self, message):
+    # type: (str) -> None
     self._connection_helper.POST('softwareInstanceBang', data={
       'computer_id': self._computer_id,
       'computer_partition_id': self.getId(),
@@ -552,17 +584,20 @@ class ComputerPartition(SlapRequester):
     return software_instance
 
   def getId(self):
+    # type: () -> str
     if not getattr(self, '_partition_id', None):
       raise ResourceNotReady()
     return self._partition_id
 
   def getInstanceGuid(self):
+    # type: () -> str
     """Return instance_guid. Raise ResourceNotReady if it doesn't exist."""
     if not getattr(self, '_instance_guid', None):
       raise ResourceNotReady()
     return self._instance_guid
 
   def getState(self):
+    # type: () -> str
     """return _requested_state. Raise ResourceNotReady if it doesn't exist."""
     if not getattr(self, '_requested_state', None):
       raise ResourceNotReady()
@@ -573,6 +608,7 @@ class ComputerPartition(SlapRequester):
     return getattr(self, '_access_status', None)
 
   def getType(self):
+    # type: () -> str
     """
     return the Software Type of the instance.
     Raise RessourceNotReady if not present.
@@ -585,9 +621,11 @@ class ComputerPartition(SlapRequester):
     return software_type
 
   def getInstanceParameterDict(self):
+    # type: () -> Dict
     return getattr(self, '_parameter_dict', None) or {}
 
   def getConnectionParameterDict(self):
+    # type: () -> Dict
     connection_dict = getattr(self, '_connection_dict', None)
     if connection_dict is None:
       # XXX Backward compatibility for older slapproxy (<= 1.0.0)
@@ -635,6 +673,7 @@ class ComputerPartition(SlapRequester):
           'slave_reference': slave_reference})
 
   def getInstanceParameter(self, key):
+    # type: (str) -> str
     parameter_dict = getattr(self, '_parameter_dict', None) or {}
     try:
       return parameter_dict[key]
@@ -642,6 +681,7 @@ class ComputerPartition(SlapRequester):
       raise NotFoundError("%s not found" % key)
 
   def getConnectionParameter(self, key):
+    # type: (str) -> str
     connection_dict = self.getConnectionParameterDict()
     try:
       return connection_dict[key]
@@ -653,6 +693,7 @@ class ComputerPartition(SlapRequester):
     self.usage = usage_log
 
   def getCertificate(self):
+    # type: () -> Dict
     xml = self._connection_helper.GET('getComputerPartitionCertificate',
             params={
                 'computer_id': self._computer_id,
@@ -691,10 +732,12 @@ class ComputerPartition(SlapRequester):
 class SlapConnectionHelper(ConnectionHelper):
 
   def getComputerInformation(self, computer_id):
+    # type: (str) -> Computer
     xml = self.GET('getComputerInformation', params={'computer_id': computer_id})
     return loads(xml)
 
   def getFullComputerInformation(self, computer_id):
+    # type: (str) -> Computer
     """
     Retrieve from SlapOS Master Computer instance containing all needed
     informations (Software Releases, Computer Partitions, ...).
@@ -713,7 +756,7 @@ class SlapConnectionHelper(ConnectionHelper):
 
     return loads(xml)
 
-getHateoasUrl_cache = {}
+getHateoasUrl_cache = {} # type: Dict[Tuple[str, Optional[str], Optional[str], Optional[str], int], str]
 @implementer(interface.slap)
 class slap:
 
@@ -722,6 +765,7 @@ class slap:
                            master_ca_file=None,
                            timeout=60,
                            slapgrid_rest_uri=None):
+    # type: (str, Optional[str], Optional[str], Optional[str], int, Optional[str]) -> None
     if master_ca_file:
       raise NotImplementedError('Master certificate not verified in this version: %s' % master_ca_file)
 
@@ -740,17 +784,17 @@ class slap:
           bytes2str(self._connection_helper.GET('getHateoasUrl'))
       except:
         pass
+    self._hateoas_navigator = None  # type: Optional[SlapHateoasNavigator]
     if slapgrid_rest_uri:
       self._hateoas_navigator = SlapHateoasNavigator(
           slapgrid_rest_uri,
           key_file, cert_file,
           master_ca_file, timeout
       )
-    else:
-      self._hateoas_navigator = None
 
   # XXX-Cedric: this method is never used and thus should be removed.
   def registerSoftwareRelease(self, software_release):
+    # type: (str) -> SoftwareRelease
     """
     Registers connected representation of software release and
     returns SoftwareRelease class object
@@ -761,6 +805,7 @@ class slap:
     )
 
   def registerToken(self):
+    # type: () -> Token
     """
     Registers connected represenation of token and
     return Token class object
@@ -773,8 +818,8 @@ class slap:
       hateoas_navigator=self._hateoas_navigator
     )
 
-
   def registerComputer(self, computer_guid):
+    # type: (str) -> Computer
     """
     Registers connected representation of computer and
     returns Computer class object
@@ -785,6 +830,7 @@ class slap:
     )
 
   def registerComputerPartition(self, computer_guid, partition_id):
+    # type: (str, str) -> ComputerPartition
     """
     Registers connected representation of computer partition and
     returns Computer Partition class object
@@ -807,12 +853,14 @@ class slap:
     return result
 
   def registerOpenOrder(self):
+    # type: () -> OpenOrder
     return OpenOrder(
       connection_helper=self._connection_helper,
       hateoas_navigator=self._hateoas_navigator
   )
 
   def registerSupply(self):
+    # type: () -> Supply
     return Supply(
       connection_helper=self._connection_helper,
       hateoas_navigator=self._hateoas_navigator
@@ -820,6 +868,7 @@ class slap:
 
   def getSoftwareReleaseListFromSoftwareProduct(self,
       software_product_reference=None, software_release_url=None):
+    # type: (Optional[str], Optional[str]) -> List[SoftwareRelease]
     url = 'getSoftwareReleaseListFromSoftwareProduct'
     params = {}
     if software_product_reference:
@@ -839,11 +888,13 @@ class slap:
     return result
 
   def getOpenOrderDict(self):
+    # XXX type: () -> Dict[str, str]
     if not getattr(self, '_hateoas_navigator', None):
       raise Exception('SlapOS Master Hateoas API required for this operation is not availble.')
     return self._hateoas_navigator.getHostingSubscriptionDict()
 
-  def getComputerDict(self): 
+  def getComputerDict(self):
+    # XXX type: () -> Dict[str, str]
     if not getattr(self, '_hateoas_navigator', None):
       raise Exception('SlapOS Master Hateoas API required for this operation is not availble.')
     return self._hateoas_navigator.getComputerDict()

slapos/slap/standalone.py

@@ -38,10 +38,16 @@ import shutil
 from six.moves import urllib
 from six.moves import http_client
 
-try:
-  import subprocess32 as subprocess
-except ImportError:
+from typing import TYPE_CHECKING, Optional, Iterable, Dict
+if TYPE_CHECKING:
   import subprocess
+  from ..slap.slap import Computer, ComputerPartition, SoftwareState, InstanceState, PartitionParameters, FilterParameters
+
+else:
+  try:
+    import subprocess32 as subprocess
+  except ImportError:
+    import subprocess
 
 import xml_marshaller
 import zope.interface
@@ -80,9 +86,11 @@ class ConfigWriter(object):
   """Base class for an object writing a config file or wrapper script.
   """
   def __init__(self, standalone_slapos):
+    # type: (StandaloneSlapOS) -> None
     self._standalone_slapos = standalone_slapos
 
   def writeConfig(self, path):
+    # type: (str) -> None
     NotImplemented
 
 
@@ -90,6 +98,7 @@ class SupervisorConfigWriter(ConfigWriter):
   """Write supervisor configuration at etc/supervisor.conf
   """
   def _getProgramConfig(self, program_name, command, stdout_logfile):
+    # type: (str, str, str) -> str
     """Format a supervisor program block.
     """
     return textwrap.dedent(
@@ -108,6 +117,7 @@ class SupervisorConfigWriter(ConfigWriter):
     """).format(**locals())
 
   def _getSupervisorConfigParts(self):
+    # type: () -> Iterable[str]
     """Iterator on parts of formatted config.
     """
     standalone_slapos = self._standalone_slapos
@@ -143,6 +153,7 @@ class SupervisorConfigWriter(ConfigWriter):
               'stdout_logfile', 'AUTO').format(self=standalone_slapos))
 
   def writeConfig(self, path):
+    # type: (str) -> None
     with open(path, 'w') as f:
       for part in self._getSupervisorConfigParts():
         f.write(part)
@@ -151,8 +162,10 @@ class SupervisorConfigWriter(ConfigWriter):
 class SlapOSConfigWriter(ConfigWriter):
   """Write slapos configuration at etc/slapos.cfg
   """
+
   def writeConfig(self, path):
-    standalone_slapos = self._standalone_slapos  # type: StandaloneSlapOS
+    # type: (str) -> None
+    standalone_slapos = self._standalone_slapos
     read_only_shared_part_list = '\n  '.join( #  pylint: disable=unused-variable; used in format()
         standalone_slapos._shared_part_list)
     with open(path, 'w') as f:
@@ -183,6 +196,7 @@ class SlapOSCommandWriter(ConfigWriter):
   """Write a bin/slapos wrapper.
   """
   def writeConfig(self, path):
+    # type: (str) -> None
     with open(path, 'w') as f:
       f.write(
           textwrap.dedent(
@@ -215,7 +229,9 @@ class StandaloneSlapOS(object):
       shared_part_list=(),
       software_root=None,
       instance_root=None,
-      shared_part_root=None):
+      shared_part_root=None,
+  ):
+    # type: (str, str, int, str, Iterable[str], Optional[str], Optional[str], Optional[str]) -> None
     """Constructor, creates a standalone slapos in `base_directory`.
 
     Arguments:
@@ -273,6 +289,7 @@ class StandaloneSlapOS(object):
     self._initBaseDirectory(software_root, instance_root, shared_part_root)
 
   def _initBaseDirectory(self, software_root, instance_root, shared_part_root):
+    # type: (Optional[str], Optional[str], Optional[str]) -> None
     """Create the directory after checking it's not too deep.
     """
     base_directory = self._base_directory
@@ -337,6 +354,7 @@ class StandaloneSlapOS(object):
 
   @property
   def computer(self):
+    # type: () -> Computer
     """Access the computer.
     """
     return self._slap.registerComputer(self._computer_id)
@@ -391,6 +409,7 @@ class StandaloneSlapOS(object):
       ipv4_address,
       ipv6_address,
       partition_base_name="slappart"):
+    # type: (int, str, str, str) -> None
     """Creates `partition_count` partitions.
 
     All partitions have the same `ipv4_address` and `ipv6_address` and
@@ -489,6 +508,7 @@ class StandaloneSlapOS(object):
         os.unlink(supervisor_conf)
 
   def supply(self, software_url, computer_guid=None, state="available"):
+    # type: (str, Optional[str], SoftwareState) -> None
     """Supply a software, see ISupply.supply
 
     Software can only be supplied on this embedded computer.
@@ -510,6 +530,7 @@ class StandaloneSlapOS(object):
       partition_parameter_kw=None,
       filter_kw=None,
       state=None):
+    # type: (str, str, Optional[str], bool, Optional[PartitionParameters], Optional[FilterParameters], Optional[InstanceState]) -> ComputerPartition
     """Request an instance, see IRequester.request
 
     Instance can only be requested on this embedded computer.
@@ -526,6 +547,7 @@ class StandaloneSlapOS(object):
         state=state)
 
   def start(self):
+    # type: () -> None
     """Start the system.
 
     If system was stopped, it will start partitions.
@@ -536,6 +558,7 @@ class StandaloneSlapOS(object):
     self._ensureSlapOSAvailable()
 
   def stop(self):
+    # type: () -> None
     """Stops all services.
 
     This methods blocks until services are stopped or a timeout is reached.
@@ -573,6 +596,7 @@ class StandaloneSlapOS(object):
               alive + instance_process_alive))
 
   def waitForSoftware(self, max_retry=0, debug=False, error_lines=30):
+    # type: (int, bool, int) -> None
     """Synchronously install or uninstall all softwares previously supplied/removed.
 
     This method retries on errors. If after `max_retry` times there's
@@ -594,6 +618,7 @@ class StandaloneSlapOS(object):
     )
 
   def waitForInstance(self, max_retry=0, debug=False, error_lines=30):
+    # type: (int, bool, int) -> None
     """Instantiate all partitions previously requested for start.
 
     This method retries on errors. If after `max_retry` times there's
@@ -615,6 +640,7 @@ class StandaloneSlapOS(object):
     )
 
   def waitForReport(self, max_retry=0, debug=False, error_lines=30):
+    # type: (int, bool, int) -> None
     """Destroy all partitions previously requested for destruction.
 
     This method retries on errors. If after `max_retry` times there's
@@ -637,17 +663,19 @@ class StandaloneSlapOS(object):
 
   def _runSlapOSCommand(
       self, command, max_retry=0, debug=False, error_lines=30):
+    # type: (str, int, bool, int) -> None
     if debug:
       prog = self._slapos_commands[command]
       # used in format(**locals()) below
       debug_args = prog.get('debug_args', '')  # pylint: disable=unused-variable
       command = prog['command'].format(**locals())
       try:
-        return subprocess.check_call(
+        subprocess.check_call(
             command,
             shell=True,
             env=self._getSubprocessEnvironment(),
         )
+        return
       except subprocess.CalledProcessError as e:
         if e.returncode == SLAPGRID_PROMISE_FAIL:
           self._logger.exception('Promise error when running %s', command)
@@ -687,6 +715,7 @@ class StandaloneSlapOS(object):
         retry += 1
 
   def _ensureSupervisordStarted(self):
+    # type: () -> None
     if os.path.exists(self._supervisor_pid):
       with open(self._supervisor_pid, 'r') as f:
         try:
@@ -715,6 +744,7 @@ class StandaloneSlapOS(object):
     self._logger.debug("Started new supervisor: %s", output)
 
   def _isSlapOSAvailable(self):
+    # type: () -> bool
     try:
       urllib.request.urlopen(self._master_url).close()
     except urllib.error.HTTPError as e:
@@ -723,6 +753,7 @@ class StandaloneSlapOS(object):
         return True
       raise
     except urllib.error.URLError as e:
+      assert isinstance(e.reason, OSError)
       if e.reason.errno == errno.ECONNREFUSED:
         return False
       raise
@@ -735,6 +766,7 @@ class StandaloneSlapOS(object):
     return True  # (if / becomes 200 OK)
 
   def _ensureSlapOSAvailable(self):
+    # type: () -> None
     # Wait for proxy to accept connections
     for i in range(2**8):
       if self._isSlapOSAvailable():
@@ -743,12 +775,14 @@ class StandaloneSlapOS(object):
     raise RuntimeError("SlapOS not started")
 
   def _getSubprocessEnvironment(self):
+    # type: () -> Optional[Dict[str, str]]
     # Running tests with `python setup.py test` sets a PYTHONPATH that
     # is suitable for current python, but problematic when this process
     # runs another version of python in subprocess.
     if 'PYTHONPATH' in os.environ:
       self._logger.warning(
-        "Removing $PYTHONPATH from environment for subprocess")
+          "Removing $PYTHONPATH from environment for subprocess")
       env = os.environ.copy()
       del env['PYTHONPATH']
       return env
+    return None

slapos/tests/test_prune.py

@@ -32,10 +32,11 @@ import shutil
 import unittest
 import slapos.client
 
-try:
-  import mock
-except ImportError:
+import sys
+if sys.version_info[0] == 3:
   from unittest import mock
+else:
+  import mock
 
 from slapos.cli.prune import do_prune
 

slapos/tests/test_slapproxy.py

@@ -34,10 +34,15 @@ import os
 import logging
 import shutil
 import socket
-try:
-  import subprocess32 as subprocess
-except ImportError:
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
   import subprocess
+else:
+  try:
+    import subprocess32 as subprocess
+  except ImportError:
+    import subprocess
+
 import sys
 import tempfile
 import time