* use annotation style without comment
* add noqa to import statement to prevent flake8 from complaining
those which import typing modules
to silence the warnings like:
```
flake8 run-test: commands[0] | flake8 --config=tox.ini alerts balancer cephadm cli_api crash devicehealth diskprediction_local hello iostat localpool nfs orchestrator prometheus selftest
cephadm/schedule.py:5:1: F401 'typing.Callable' imported but unused
cephadm/schedule.py:8:1: F401 'ceph.deployment.service_spec.ServiceSpec' imported but unused
cephadm/tests/fixtures.py:17:1: F401 'orchestrator.OrchResult' imported but unused
orchestrator/module.py:4:1: F401 'typing.Set' imported but unused
orchestrator/module.py:17:1: F401 'ceph.deployment.inventory.Device' imported but unused
prometheus/module.py:17:1: F401 'typing.DefaultDict' imported but unused
6 F401 'typing.Callable' imported but unused
ERROR: InvocationError for command /home/jenkins-build/build/workspace/ceph-pull-requests/src/pybind/mgr/.tox/flake8/bin/flake8 --config=tox.ini alerts
```
Signed-off-by: Kefu Chai <tchaikov@gmail.com>
(cherry picked from commit
7894e7bdd1c25d86a71e3dcdc88c46062cb31101)
Conflicts:
src/pybind/mgr/orchestrator/module.py [ commit
28314cebef32
("mgr/cephadm: Adding support for natural sorting") not in
pacific ]
src/pybind/mgr/cephadm/services/exporter.py [ additional
pacific-only change, this file doesn't exist in main ]
class HostAssignment(object):
def __init__(self,
- spec, # type: ServiceSpec
+ spec: ServiceSpec,
hosts: List[orchestrator.HostSpec],
unreachable_hosts: List[orchestrator.HostSpec],
daemons: List[orchestrator.DaemonDescription],
networks: Dict[str, Dict[str, Dict[str, List[str]]]] = {},
- filter_new_host=None, # type: Optional[Callable[[str],bool]]
+ filter_new_host: Optional[Callable[[str], bool]] = None,
allow_colo: bool = False,
primary_daemon_type: Optional[str] = None,
per_host_daemon_type: Optional[str] = None,
required_keys = ['crt', 'key', 'token', 'port']
DEFAULT_PORT = '9443'
- def __init__(self, mgr, crt="", key="", token="", port=""):
- # type: (CephadmOrchestrator, str, str, str, str) -> None
+ def __init__(self, mgr: "CephadmOrchestrator", crt: str = "", key: str = "",
+ token: str = "", port: str = "") -> None:
self.mgr = mgr
self.crt = crt
self.key = key
yield m
-def wait(m, c):
- # type: (CephadmOrchestrator, OrchResult) -> Any
+def wait(m: CephadmOrchestrator, c: OrchResult) -> Any:
return raise_if_exception(c)
import yaml
from prettytable import PrettyTable
-from ceph.deployment.inventory import Device
+from ceph.deployment.inventory import Device # noqa: F401; pylint: disable=unused-variable
from ceph.deployment.drive_group import DriveGroupSpec, DeviceSelection, OSDMethod
from ceph.deployment.service_spec import PlacementSpec, ServiceSpec, service_spec_allow_invalid_from_json
from ceph.deployment.hostspec import SpecValidationError
def __init__(self, *args: Any, **kwargs: Any) -> None:
super(OrchestratorCli, self).__init__(*args, **kwargs)
- self.ident = set() # type: Set[str]
- self.fault = set() # type: Set[str]
+ self.ident: Set[str] = set()
+ self.fault: Set[str] = set()
self._load()
self._refresh_health()
pg_summary = self.get('pg_summary')
for pool in pg_summary['by_pool']:
- num_by_state = defaultdict(int) # type: DefaultDict[str, int]
+ num_by_state: DefaultDict[str, int] = defaultdict(int)
for state_name, count in pg_summary['by_pool'][pool].items():
for state in state_name.split('+'):