使用获取的jobconfs来创建扫描任务
(project, job_confs, creator=None, puppy_create=False, **kwargs)
| 142 | |
| 143 | |
| 144 | def create_job(project, job_confs, creator=None, puppy_create=False, **kwargs): |
| 145 | """ |
| 146 | 使用获取的jobconfs来创建扫描任务 |
| 147 | """ |
| 148 | job_manager = JobManager(project) |
| 149 | job_context = job_confs["job_context"] |
| 150 | incr_scan = job_context.get("incr_scan", True) |
| 151 | force_create = job_confs.get("force_create") or job_context.get("force_create", False) |
| 152 | created_from = job_confs.get("created_from") or job_context.get("created_from") or "codedog_web" |
| 153 | job_manager.check_job_scm_url(job_context) |
| 154 | job = job_manager.initialize_job(force_create, creator, created_from) |
| 155 | job = job_manager.update_job(job, job_confs) |
| 156 | if kwargs.get("scan_type") is not None: |
| 157 | scan_type = kwargs["scan_type"] |
| 158 | else: |
| 159 | scan_type = ScanTypeEnum.INCRESE if incr_scan else ScanTypeEnum.FULL |
| 160 | scan = job_manager.create_scan_on_analysis_server(job, job_context, scan_type) |
| 161 | job_context.update({"scan_id": scan["id"]}) |
| 162 | job_manager.insert_auth_into_job_context(job_context) |
| 163 | job_manager.create_tasks(job, job_confs, puppy_create) |
| 164 | return job.id, scan["id"] |
| 165 | |
| 166 | |
| 167 | def finish_job_from_client(job, project, job_confs, puppy_create=False, **kwargs): |
nothing calls this directly
no test coverage detected