class Scan:
    # Scan Permissions
    PERMISSION_NO_ACCESS = 0
    PERMISSION_CAN_VIEW = 16
    PERMISSION_CAN_CONTROL = 32
    PERMISSION_CAN_CONFIGURE = 64 class ScanHostDetails(BaseModel):

    def __init__(
            self,
            info=None,
            vulnerabilities=None,
            compliance=None,
    ):
        self._info = None
        self._vulnerabilities = None
        self._compliance = None

        self.info = info
        self.vulnerabilities = vulnerabilities
        self.compliance = compliance

    @property
    def info(self):
        return self._info

    @info.setter
    @BaseModel._model(ScanHostInfo)
    def info(self, info):
        self._info = info

    @property
    def compliance(self):
        return self._compliance

    @compliance.setter
    @BaseModel._model_list(ScanHostCompliance)
    def compliance(self, compliance):
        self._compliance = compliance

    @property
    def vulnerabilities(self):
        return self._vulnerabilities

    @vulnerabilities.setter
    @BaseModel._model_list(ScanHostVulnerability)
    def vulnerabilities(self, vulnerabilities):
        self._vulnerabilities = vulnerabilities class ScanList(BaseModel):

    def __init__(
            self,
            folders=None,
            scans=None,
            timestamp=None,
    ):
        self._scans = None
        self.folders = folders
        self.scans = scans
        self.timestamp = timestamp

    @property
    def scans(self):
        return self._scans

    @scans.setter
    @BaseModel._model_list(Scan)
    def scans(self, scans):
        self._scans = scans class ScanSettings(BaseModel):

    def __init__(
            self,
            name,
            text_targets=None,
            description=None,
            emails=None,
            enabled=False,
            launch=None,
            starttime=None,
            rrules=None,
            timezone=None,
            file_targets=None,
            launch_now=None,
            folder_id=None,
            policy_id=None,
            scanner_id=None,
            acls=[],
            asset_lists=[],
            agent_group_id=[],
            tag_targets=[],
    ):
        self.name = name
        self.description = description
        self.emails = emails
        self.enabled = enabled
        self.launch = launch
        self.starttime = starttime
        self.rrules = rrules
        self.timezone = timezone
        self.file_targets = file_targets
        self.launch_now = launch_now
        self.folder_id = folder_id
        self.policy_id = policy_id
        self.scanner_id = scanner_id
        self.text_targets = text_targets
        self.acls = acls
        self.asset_lists = asset_lists
        self.agent_group_id = agent_group_id
        self.tag_targets = tag_targets class ScanDetails(BaseModel):

    def __init__(
            self,
            info=None,
            hosts=None,
            comphosts=None,
            notes=None,
            remediations=None,
            vulnerabilities=None,
            compliance=None,
            history=None,
            filters=None,
    ):
        self._info = None
        self._history = None
        self._hosts = None

        self.info = info
        self.hosts = hosts
        self.comphosts = comphosts
        self.notes = notes
        self.remediations = remediations
        self.vulnerabilities = vulnerabilities
        self.compliance = compliance
        self.history = history
        self.filters = filters

    @property
    def info(self):
        return self._info

    @info.setter
    @BaseModel._model(ScanInfo)
    def info(self, info):
        self._info = info

    @property
    def history(self):
        return self._history

    @history.setter
    @BaseModel._model_list(ScanDetailsHistory)
    def history(self, history):
        self._history = history

    @property
    def hosts(self):
        return self._hosts

    @hosts.setter
    @BaseModel._model_list(ScanHost)
    def hosts(self, hosts):
        self._hosts = hosts class ScansApi(BaseApi):

    STATUS_EXPORT_READY = u'ready'

    def configure(self, scan_id=None, scan_configure=None, schedule_uuid=None):
        """Configure an existing scan.

        :param scan_id: The scan ID.
        :param scan_configure: An instance of :class:`ScanConfigureRequest`.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: The ID of scan just configured.
        """
        response = self._client.put('scans/%(scan_id)s',
                                    scan_configure,
                                    path_params={'scan_id': scan_id or schedule_uuid})
        return loads(response.text).get('scan', {}).get('id')

    def create(self, scan_create, return_uuid=False):
        """Create a scan.

        :param scan_create: An instance of :class:`ScanCreateRequest`.
        :param return_uuid: Optional param to return uuid rather than numeric id as the method response.
        :raise TenableIOApiException:  When API error is encountered.
        :return: The ID of scan just created.
        """
        response = self._client.post('scans', scan_create)
        return loads(response.text).get('scan', {}).get('id') \
            if not return_uuid \
            else loads(response.text).get('scan', {}).get('uuid')

    def copy(self, scan_id=None, schedule_uuid=None):
        """Creates a copy of a scan.

        :param scan_id: The scan ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :return: An instance of :class:`tenable_io.api.models.Scan`.
        """
        response = self._client.post('scans/%(scan_id)s/copy',
                                     {},
                                     path_params={'scan_id': scan_id or schedule_uuid})
        return Scan.from_json(response.text)

    def delete(self, scan_id=None, schedule_uuid=None):
        """Delete a scan. NOTE: Scans in running, paused or stopping states can not be deleted.

        :raise TenableIOApiException:  When API error is encountered.
        :param scan_id: The scan ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :return: True if successful.
        """
        self._client.delete('scans/%(scan_id)s', path_params={'scan_id': scan_id or schedule_uuid})
        return True

    def details(self, scan_id=None, history_id=None, schedule_uuid=None):
        """Return details of the given scan.

        :param scan_id: The scan ID.
        :param history_id: The historical data ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: An instance of :class:`tenable_io.api.models.ScanDetails`.
        """
        response = self._client.get('scans/%(scan_id)s',
                                    path_params={'scan_id': scan_id or schedule_uuid},
                                    params={'history_id': history_id} if history_id else None)

        return ScanDetails.from_json(response.text)

    def export_download(self, scan_id=None, file_id=None, stream=True, chunk_size=1024, is_was=False, schedule_uuid=None):
        """Download an exported scan.

        :param scan_id: The scan ID.
        :param file_id: The file ID.
        :param stream: Default to True. If False, the response content will be immediately downloaded.
        :param chunk_size: If Stream=False, data is returned as a single chunk.\
         If Stream=True, it's the number of bytes it should read into memory.
        :param is_was: A flag that specifies that the scan is a WAS type scan.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: The downloaded file.
        """
        response = self._client.get('scans/%(scan_id)s/export/%(file_id)s/download',
                                    path_params={'scan_id': scan_id or schedule_uuid, 'file_id': file_id},
                                    params={'type': ScanExportRequest.WAS_EXPORT_TYPE} if is_was else None,
                                    stream=stream)
        return response.iter_content(chunk_size=chunk_size)

    def export_request(self, scan_id=None, scan_export=None, history_id=None, is_was=False, schedule_uuid=None):
        """Export the given scan. Once requested, the file can be downloaded using the export\
         download method upon receiving a "ready" status from the export status method.

        :param scan_id: The scan ID.
        :param scan_export: An instance of :class:`ScanExportRequest`.
        :param history_id: The history ID of historical data.
        :param is_was: A flag that specifies that the scan is a WAS type scan.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: The file ID.
        """
        assert isinstance(scan_export, ScanExportRequest)
        params = {'history_id': history_id if history_id else None,
                  'type': ScanExportRequest.WAS_EXPORT_TYPE if is_was else None}
        response = self._client.post('scans/%(scan_id)s/export',
                                     scan_export,
                                     path_params={'scan_id': scan_id or schedule_uuid},
                                     params={k: v for k, v in params.items() if v is not None})
        return loads(response.text).get('file')

    def export_status(self, scan_id=None, file_id=None, is_was=False, schedule_uuid=None):
        """Check the file status of an exported scan. When an export has been requested,\
         it is necessary to poll this endpoint until a "ready" status is returned,\
          at which point the file is complete and can be downloaded using the export download endpoint.

        :param scan_id: The scan ID.
        :param file_id: The file ID.
        :param is_was: A flag that specifies that the scan is a WAS type scan.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: The file status.
        """
        response = self._client.get('scans/%(scan_id)s/export/%(file_id)s/status',
                                    path_params={'scan_id': scan_id or schedule_uuid, 'file_id': file_id},
                                    params={'type': ScanExportRequest.WAS_EXPORT_TYPE} if is_was else None)
        return loads(response.text).get('status')

    def folder(self, scan_id=None, folder_id=None, schedule_uuid=None):
        """Move to a scan to a folder.

        :param scan_id: The scan ID.
        :param folder_id: The folder ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: True if successful.
        """
        self._client.put('scans/%(scan_id)s/folder',
                         {'folder_id': folder_id},
                         path_params={'scan_id': scan_id or schedule_uuid})
        return True

    def history(self, scan_id=None, history_id=None, schedule_uuid=None):
        """Returns a scan history.

        :param scan_id: The scan ID.
        :param history_id: The historical data ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :return: An instance of :class:`tenable_io.api.models.ScanHistory`.
        """
        response = self._client.get('scans/%(scan_id)s/history/%(history_id)s',
                                    path_params={'scan_id': scan_id or schedule_uuid, 'history_id': history_id})
        return ScanHistory.from_json(response.text)

    def host_details(self, scan_id=None, host_id=None, schedule_uuid=None):
        """Returns details for the given host.

        :param scan_id: The scan ID.
        :param host_id: The host ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: An instance of :class:`tenable_io.api.models.ScanHostDetails`.
        """
        response = self._client.get('scans/%(scan_id)s/hosts/%(host_id)s',
                                    path_params={'scan_id': scan_id or schedule_uuid, 'host_id': host_id})
        return ScanHostDetails.from_json(response.text)

    def import_scan(self, scan_import, include_aggregate=True):
        """Import an existing scan which has been uploaded using :func:`TenableIO.FileApi.upload`

        :param scan_import: An instance of :class:`ScanImportRequest`.
        :param include_aggregate: Boolean indicating whether scan data should appear in Workbenches.
        :raise TenableIOApiException:  When API error is encountered.
        :return: The ID of the imported scan.
        """
        aggregate_option = 0
        if include_aggregate:
            aggregate_option = 1

        response = self._client.post('scans/import?include_aggregate=%(include_aggregate)s',
                                     scan_import,
                                     path_params={'include_aggregate': aggregate_option})
        return loads(response.text).get('scan', {}).get('id')

    def launch(self, scan_id=None, scan_launch_request=None, schedule_uuid=None):
        """Launch a scan.

        :param scan_id: The scan ID.
        :param scan_launch_request: An instance of :class:`ScanLaunchRequest`.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: The scan uuid.
        """
        assert isinstance(scan_launch_request, ScanLaunchRequest)
        response = self._client.post('scans/%(scan_id)s/launch',
                                     scan_launch_request,
                                     path_params={'scan_id': scan_id or schedule_uuid})
        return loads(response.text).get('scan_uuid')

    def list(self, folder_id=None, last_modification_date=None):
        """Return the scan list.

        :param folder_id: The folder ID (optional).
        :param last_modification_date: Limit the results to those that have only changed since this time (optional).
        :raise TenableIOApiException:  When API error is encountered.
        :return: An instance of :class:`tenable_io.api.models.ScanList`.
        """
        params = {'folder_id': folder_id, 'last_modification_date': last_modification_date}
        response = self._client.get('scans', params={k: v for (k, v) in params.items() if v})
        return ScanList.from_json(response.text)

    def pause(self, scan_id=None, schedule_uuid=None):
        """Pause a scan.

        :param scan_id: The scan ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: True if successful.
        """
        self._client.post('scans/%(scan_id)s/pause', {}, path_params={'scan_id': scan_id or schedule_uuid})
        return True

    def resume(self, scan_id=None, schedule_uuid=None):
        """Resume a scan.

        :param scan_id: The scan ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: True if successful.
        """
        self._client.post('scans/%(scan_id)s/resume', {}, path_params={'scan_id': scan_id or schedule_uuid})
        return True

    def stop(self, scan_id=None, schedule_uuid=None):
        """Stop a scan.

        :param scan_id: The scan ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: True if successful.
        """
        self._client.post('scans/%(scan_id)s/stop', {}, path_params={'scan_id': scan_id or schedule_uuid})
        return True

    def latest_status(self, scan_id=None, schedule_uuid=None):
        """ Gets scan latest status.

        :param scan_id: The scan ID.
        :param schedule_uuid: The scan schedule UUID, this value will be used when specified and when scan_id is not present.
        :raise TenableIOApiException:  When API error is encountered.
        :return: the scan status.
        """
        response = self._client.get('scans/%(scan_id)s/latest-status',
                                    path_params={'scan_id': scan_id or schedule_uuid})
        return loads(response.text).get('status', '') class ScanConfigureRequest(ScanSaveRequest):

    def __init__(
            self,
            uuid=None,
            settings=None,
            credentials=None,
    ):
        super(ScanConfigureRequest, self).__init__(uuid, settings, credentials) class ScanCreateRequest(ScanSaveRequest):

    def __init__(
            self,
            uuid,
            settings=None,
            credentials=None,
    ):
        super(ScanCreateRequest, self).__init__(uuid, settings, credentials) class ScanExportRequest(BaseRequest):

    CHAPTER_COMPLIANCE = u'compliance'
    CHAPTER_COMPLIANCE_EXEC = u'compliance_exec'
    CHAPTER_CUSTOM_VULN_BY_HOST = u'vuln_by_host'
    CHAPTER_CUSTOM_VULN_BY_PLUGIN = u'vuln_by_plugin'
    CHAPTER_EXECUTIVE_SUMMARY = u'vuln_hosts_summary'
    CHAPTER_REMEDIATIONS = u'remediations'

    FORMAT_CSV = u'csv'
    FORMAT_DB = u'db'
    FORMAT_HTML = u'html'
    FORMAT_NESSUS = u'nessus'
    FORMAT_PDF = u'pdf'

    WAS_EXPORT_TYPE = u'web-app'

    def __init__(
            self,
            format,
            password=None,
            chapters=None,
    ):
        assert format in [
            ScanExportRequest.FORMAT_CSV,
            ScanExportRequest.FORMAT_DB,
            ScanExportRequest.FORMAT_HTML,
            ScanExportRequest.FORMAT_NESSUS,
            ScanExportRequest.FORMAT_PDF,
        ]
        assert chapters in [
            None,
            ScanExportRequest.CHAPTER_COMPLIANCE,
            ScanExportRequest.CHAPTER_COMPLIANCE_EXEC,
            ScanExportRequest.CHAPTER_CUSTOM_VULN_BY_HOST,
            ScanExportRequest.CHAPTER_CUSTOM_VULN_BY_PLUGIN,
            ScanExportRequest.CHAPTER_EXECUTIVE_SUMMARY,
            ScanExportRequest.CHAPTER_REMEDIATIONS,
        ]
        self.format = format
        self.password = password
        self.chapters = chapters

    def as_payload(self, filter_=None):
        return super(ScanExportRequest, self).as_payload(True) class ScanImportRequest(BaseRequest):

    def __init__(
            self,
            file,
            folder_id=None,
            password=None
    ):
        self.file = file
        self.folder_id = folder_id
        self.password = password class ScanLaunchRequest(BaseRequest):

    def __init__(
            self,
            alt_targets=None
    ):
        self.alt_targets = alt_targets class TenableIOTestConfig(object):

    @staticmethod
    def get(key):
        return config.get('tenable_io-test', key) def wait_until(expression, condition):
    value = expression()
    start_time = time()
    elapsed = 0
    while not condition(value) and elapsed <= WAIT_TIMEOUT:
        sleep(WAIT_INTERVAL)
        value = expression()
        elapsed = time() - start_time
    assert condition(value), u'Timeout waiting for a condition (%d seconds elapsed).' % elapsed
    return value
import os
import pytest
from tenable_io.api.models import Scan, ScanHostDetails, ScanList, ScanSettings, ScanDetails
from tenable_io.api.scans import ScansApi, ScanConfigureRequest, ScanCreateRequest, \
    ScanExportRequest, ScanImportRequest, ScanLaunchRequest
from tests.config import TenableIOTestConfig
from tests.integration.api.utils.utils import wait_until

@pytest.mark.extended_testing
@pytest.mark.vcr()
def test_scans_control_endpoints_using_schedule_uuid(client, new_scan):
    scan_id = new_scan
    scan_details = client.scans_api.details(schedule_uuid=new_scan)
    assert scan_details.info.status in [Scan.STATUS_CANCELED, Scan.STATUS_COMPLETED, Scan.STATUS_EMPTY], \
        u'Scan is not in an idling state.'

    # Launch the scan.
    client.scans_api.launch(
        schedule_uuid=new_scan,
        scan_launch_request=ScanLaunchRequest()
    )
    scan_details = client.scans_api.details(schedule_uuid=new_scan)
    assert scan_details.info.status in [Scan.STATUS_PENDING, Scan.STATUS_INITIALIZING, Scan.STATUS_RUNNING], \
        u'The scan is not launching.'

    scan_details = wait_until(lambda: client.scans_api.details(schedule_uuid=new_scan),
                              lambda details: details.info.status in [
                                       Scan.STATUS_COMPLETED, Scan.STATUS_RUNNING])
    assert scan_details.info.status == Scan.STATUS_RUNNING, u'The scan is not running.'

    # Pause the running scan.
    client.scans_api.pause(schedule_uuid=new_scan)
    scan_details = client.scans_api.details(schedule_uuid=new_scan)
    assert scan_details.info.status in [Scan.STATUS_PAUSED, Scan.STATUS_PAUSING], u'The scan is not pausing.'
    scan_details = wait_until(lambda: client.scans_api.details(schedule_uuid=new_scan),
                              lambda details: details.info.status in [Scan.STATUS_PAUSED])
    assert scan_details.info.status == Scan.STATUS_PAUSED, u'The scan is not paused.'

    # Resume the paused scan.
    client.scans_api.resume(schedule_uuid=new_scan)
    scan_details = client.scans_api.details(schedule_uuid=new_scan)
    assert scan_details.info.status in [Scan.STATUS_RESUMING, Scan.STATUS_RUNNING], u'The scan is not resuming.'

    scan_details = wait_until(lambda: client.scans_api.details(schedule_uuid=new_scan),
                              lambda details: details.info.status in [
                                       Scan.STATUS_COMPLETED, Scan.STATUS_RUNNING])
    assert scan_details.info.status == Scan.STATUS_RUNNING, u'The scan is not running.'

    # Stop the running scan.
    client.scans_api.stop(schedule_uuid=new_scan)
    scan_details = client.scans_api.details(schedule_uuid=new_scan)
    assert scan_details.info.status in [Scan.STATUS_CANCELED, Scan.STATUS_STOPPING], u'The scan is not resuming.'

    scan_details = wait_until(lambda: client.scans_api.details(schedule_uuid=new_scan),
                              lambda details: details.info.status in [Scan.STATUS_CANCELED])
    assert scan_details.info.status == Scan.STATUS_CANCELED, u'The scan was not canceled.'

    # Test the history API.
    assert len(scan_details.history) > 0, u'The scan should have at least one history.'
    history_id = scan_details.history[0].history_id
    history = client.scans_api.history(schedule_uuid=new_scan, history_id=history_id)
    assert history.status == scan_details.info.status, u'Scan history should report same status as scan details.'

    # Test the scan host details API.
    assert len(scan_details.hosts) > 0, u'Expected the scan to have at least one host.'
    host_details = client.scans_api.host_details(schedule_uuid=new_scan, host_id=scan_details.hosts[0].host_id)
    assert isinstance(host_details, ScanHostDetails), \
        u'The `host_details` method did not return type `ScanHostDetails`.'

@pytest.mark.extended_testing
@pytest.mark.vcr()
def test_scans_control_endpoints(client, new_scan_id):
    scan_id = new_scan_id
    scan_details = client.scans_api.details(new_scan_id)
    assert scan_details.info.status in [Scan.STATUS_CANCELED, Scan.STATUS_COMPLETED, Scan.STATUS_EMPTY], \
        u'Scan is not in an idling state.'

    # Launch the scan.
    client.scans_api.launch(
        new_scan_id,
        scan_launch_request=ScanLaunchRequest()
    )
    scan_details = client.scans_api.details(new_scan_id)
    assert scan_details.info.status in [Scan.STATUS_PENDING, Scan.STATUS_INITIALIZING, Scan.STATUS_RUNNING], \
        u'The scan is not launching.'

    scan_details = wait_until(lambda: client.scans_api.details(new_scan_id),
                              lambda details: details.info.status in [
                                       Scan.STATUS_COMPLETED, Scan.STATUS_RUNNING])
    assert scan_details.info.status == Scan.STATUS_RUNNING, u'The scan is not running.'

    # Pause the running scan.
    client.scans_api.pause(new_scan_id)
    scan_details = client.scans_api.details(new_scan_id)
    assert scan_details.info.status in [Scan.STATUS_PAUSED, Scan.STATUS_PAUSING], u'The scan is not pausing.'
    scan_details = wait_until(lambda: client.scans_api.details(new_scan_id),
                              lambda details: details.info.status in [Scan.STATUS_PAUSED])
    assert scan_details.info.status == Scan.STATUS_PAUSED, u'The scan is not paused.'

    # Resume the paused scan.
    client.scans_api.resume(new_scan_id)
    scan_details = client.scans_api.details(new_scan_id)
    assert scan_details.info.status in [Scan.STATUS_RESUMING, Scan.STATUS_RUNNING], u'The scan is not resuming.'

    scan_details = wait_until(lambda: client.scans_api.details(new_scan_id),
                              lambda details: details.info.status in [
                                       Scan.STATUS_COMPLETED, Scan.STATUS_RUNNING])
    assert scan_details.info.status == Scan.STATUS_RUNNING, u'The scan is not running.'

    # Stop the running scan.
    client.scans_api.stop(new_scan_id)
    scan_details = client.scans_api.details(new_scan_id)
    assert scan_details.info.status in [Scan.STATUS_CANCELED, Scan.STATUS_STOPPING], u'The scan is not resuming.'

    scan_details = wait_until(lambda: client.scans_api.details(new_scan_id),
                              lambda details: details.info.status in [Scan.STATUS_CANCELED])
    assert scan_details.info.status == Scan.STATUS_CANCELED, u'The scan was not canceled.'

    # Test the history API.
    assert len(scan_details.history) > 0, u'The scan should have at least one history.'
    history_id = scan_details.history[0].history_id
    history = client.scans_api.history(new_scan_id, history_id=history_id)
    assert history.status == scan_details.info.status, u'Scan history should report same status as scan details.'

    # Test the scan host details API.
    assert len(scan_details.hosts) > 0, u'Expected the scan to have at least one host.'
    host_details = client.scans_api.host_details(new_scan_id, host_id=scan_details.hosts[0].host_id)

    assert isinstance(host_details, ScanHostDetails), \