Coverage for plugins/module_utils/attachment.py : 24%

# -*- coding: utf-8 -*- 

# Copyright: (c) 2021, XLAB Steampunk <steampunk@xlab.si> 

# 

# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) 

from __future__ import absolute_import, division, print_function 

__metaclass__ = type 

import collections 

import mimetypes 

import os 

from . import errors 

def _path(api_path, *subpaths): 

return "/".join(api_path + ("attachment",) + subpaths) 

class AttachmentClient: 

def __init__(self, client, batch_size=10000): 

# 10000 records is default batch size for ServiceNow Attachment REST API, so we also use it 

# as a default. 

self.client = client 

self.batch_size = batch_size 

def list_records(self, query=None): 

base_query = dict(query or {}, sysparm_limit=self.batch_size) 

offset = 0 

total = 1 # Dummy value that ensures loop executes at least once 

result = [] 

while offset < total: 

response = self.client.get( 

_path(self.client.api_path), 

query=dict(base_query, sysparm_offset=offset), 

) 

result.extend(response.json["result"]) 

total = int(response.headers["x-total-count"]) 

offset += self.batch_size 

return result 

def create_record(self, query, data, mime_type, check_mode): 

if check_mode: 

return query 

return self.client.request( 

"POST", 

_path(self.client.api_path, "file"), 

query=query, 

headers={"Accept": "application/json", "Content-type": mime_type}, 

bytes=data, 

).json["result"] 

def upload_record(self, table, table_sys_id, metadata, check_mode): 

# Table and table_sys_id parameters uniquely identify the record we will attach a file to. 

query = dict( 

table_name=table, 

table_sys_id=table_sys_id, 

file_name=metadata["name"], 

content_type=metadata["type"], 

hash=metadata["hash"], 

) 

try: 

with open(metadata["path"], "rb") as file_obj: 

return self.create_record( 

query, file_obj.read(), query["content_type"], check_mode 

) 

except (IOError, OSError): 

raise errors.ServiceNowError("Cannot open {0}".format(metadata["path"])) 

def upload_records(self, table, table_sys_id, metadata_dict, check_mode): 

return [ 

self.upload_record( 

table, table_sys_id, dict(metadata, name=name), check_mode 

) 

for name, metadata in metadata_dict.items() 

] 

def delete_record(self, record, check_mode): 

if not check_mode: 

self.client.delete(_path(self.client.api_path, record["sys_id"])) 

def delete_attached_records(self, table, table_sys_id, check_mode): 

for record in self.list_records( 

dict(table_name=table, table_sys_id=table_sys_id) 

): 

self.delete_record(record, check_mode) 

def update_records(self, table, table_sys_id, metadata_dict, records, check_mode): 

mapped_records = dict((r["file_name"], r) for r in records) 

for name, metadata in metadata_dict.items(): 

record = mapped_records.get(name, {}) 

if record.get("hash") != metadata["hash"]: 

if record: 

self.delete_record(record, check_mode) 

mapped_records[name] = self.upload_record( 

table, table_sys_id, dict(metadata, name=name), check_mode 

) 

return list(mapped_records.values()) 

def get_attachment(self, attachment_sys_id): 

return self.client.get(_path(self.client.api_path, attachment_sys_id, "file")) 

def save_attachment(self, binary_data, dest): 

try: 

with open(str(dest), "wb") as f: 

f.write(binary_data) 

except (IOError, OSError) as e: 

raise errors.ServiceNowError(str(e)) 

def transform_metadata_list(metadata_list, hashing_method): 

metadata_dict = dict() 

dups = collections.defaultdict(list) 

for metadata in metadata_list or []: 

name = get_file_name(metadata) 

dups[name].append(metadata["path"]) 

metadata_dict[name] = { 

"path": metadata["path"], 

"type": get_file_type(metadata), 

"hash": hashing_method(metadata["path"]), 

} 

dup_sets = ["({0})".format(", ".join(v)) for v in dups.values() if len(v) > 1] 

if dup_sets: 

raise errors.ServiceNowError( 

"Found the following duplicates: {0}".format(" ".join(dup_sets)) 

) 

return metadata_dict 

def get_file_name(metadata): 

if "name" in metadata and metadata["name"] is not None: 

return metadata["name"] 

return os.path.basename(metadata["path"]) 

def get_file_type(metadata): 

if "type" in metadata and metadata["type"] is not None: 

return metadata["type"] 

return mimetypes.guess_type(metadata["path"])[0] 

def are_changed(records, metadata_dict): 

mapped_records = dict((r["file_name"], r) for r in records) 

return [ 

metadata["hash"] != mapped_records.get(name, {}).get("hash") 

for name, metadata in metadata_dict.items() 

] 

def are_changed_return_records(records, metadata_dict): 

mapped_records = dict((r["file_name"], r) for r in records) 

update = {} # filtered metadata_dict of the attachments to be added/updated 

changed = [] 

unchanged = [] 

for name, metadata in metadata_dict.items(): 

if metadata["hash"] == mapped_records.get(name, {}).get("hash"): 

unchanged.append(mapped_records[name]) 

elif mapped_records.get(name, {}): 

# if record with the same file_name already exists 

changed.append(mapped_records[name]) 

update[name] = metadata 

else: 

update[name] = metadata 

return update, changed, unchanged