Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

style: format code with Black and PHP CS Fixer #571

Merged
merged 1 commit into from
Dec 1, 2023
Merged

style: format code with Black and PHP CS Fixer #571

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 44 additions & 19 deletions services/mmc/plugins/xmppmaster/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1125,25 +1125,36 @@ def xmpp_get_info_synchro_packageid(uuidpackage):

def get_agent_descriptor_base():
# Sending IQ request and data reception
data = ObjectXmpp().iqsendpulse('rspulse@pulse/mainrelay',
{"action": "remotexmppmonitoring", "data": "agentinfos"}, 300)
data = ObjectXmpp().iqsendpulse(
"rspulse@pulse/mainrelay",
{"action": "remotexmppmonitoring", "data": "agentinfos"},
300,
)

result = json.loads(data)

if 'result' in result:
if "result" in result:
resultdata = zlib.decompress(base64.b64decode(result["result"]))
resultdata_str = resultdata.decode('utf-8')
resultdata_str = resultdata.decode("utf-8")
resultdata = json.loads(resultdata_str)

logger.error("Voilà resultdata dans get_agent_descriptor_base %s" % str(resultdata))
logger.error(
"Voilà resultdata dans get_agent_descriptor_base %s" % str(resultdata)
)

agent_descriptor = resultdata.get("agentdescriptor", "{}")
pathagent = resultdata.get("pathagent", "")

if isinstance(agent_descriptor, str):
agent_descriptor = json.loads(agent_descriptor)

lib_agent_sorted = {k: agent_descriptor['lib_agent'][k] for k in sorted(agent_descriptor.get("lib_agent", {}), key=lambda x: (x.startswith('__'), x))}
lib_agent_sorted = {
k: agent_descriptor["lib_agent"][k]
for k in sorted(
agent_descriptor.get("lib_agent", {}),
key=lambda x: (x.startswith("__"), x),
)
}

new_structure = {
"autoupdate": True,
Expand All @@ -1155,66 +1166,80 @@ def get_agent_descriptor_base():
"script_agent": agent_descriptor.get("script_agent", {}),
"fingerprint": agent_descriptor.get("fingerprint", ""),
},
"dir_agent_base": pathagent
"dir_agent_base": pathagent,
}

return new_structure
else:
logger.error("The 'Result' key is missing in the result. Check the logs of the agent.")
logger.error(
"The 'Result' key is missing in the result. Check the logs of the agent."
)
return {}


def get_plugin_lists():
base_plugin_path = os.path.join("/","var","lib","pulse2","xmpp_baseplugin")
base_plugin_path = os.path.join("/", "var", "lib", "pulse2", "xmpp_baseplugin")
files = []
_files = []
plugins = {}
for _, dir, _files in os.walk(base_plugin_path):
if dir != "__pycache__":
# remove "plugin_" and ".py" from the name
files = [file[7:-3] for file in _files if file.endswith('.py')]
files = [file[7:-3] for file in _files if file.endswith(".py")]
break

for file in files:
meta = {}

with open(os.path.join(base_plugin_path, "plugin_%s.py"%file), 'r') as plugin_fb:
with open(
os.path.join(base_plugin_path, "plugin_%s.py" % file), "r"
) as plugin_fb:
line = ""
while line.startswith("plugin = ") is False:
line = plugin_fb.readline()
line = line.split('#')[0]
line = line.split("#")[0]
try:
meta = json.loads(line.replace("plugin =", ""))
except:
pass
plugins[file] = [meta['VERSION'], meta['TYPE'], meta['VERSIONAGENT'] if 'VERSIONAGENT' in meta else "0.0.0"]
plugins[file] = [
meta["VERSION"],
meta["TYPE"],
meta["VERSIONAGENT"] if "VERSIONAGENT" in meta else "0.0.0",
]
plugin_fb.close()

base_pluginscheduler_path = os.path.join("/","var","lib","pulse2","xmpp_basepluginscheduler")
base_pluginscheduler_path = os.path.join(
"/", "var", "lib", "pulse2", "xmpp_basepluginscheduler"
)
pluginsscheduled = {}
for _, dir, _files in os.walk(base_pluginscheduler_path):
if dir != "__pycache__":
# remove "plugin_" and ".py" from the name
files = [file[0:-3] for file in _files if file.endswith('.py')]
files = [file[0:-3] for file in _files if file.endswith(".py")]
break

for file in files:
meta = {}
with open(os.path.join(base_pluginscheduler_path, "%s.py"%file), 'r') as plugin_fb:
with open(
os.path.join(base_pluginscheduler_path, "%s.py" % file), "r"
) as plugin_fb:
line = ""
while line.startswith("plugin = ") is False:
line = plugin_fb.readline()
line = line.split('#')[0]
line = line.split("#")[0]
try:
meta = json.loads(line.replace("plugin = ", ""))
except Exception as e:
meta = eval(line.replace("plugin = ", ""))

pluginsscheduled[file] = meta['VERSION']
pluginsscheduled[file] = meta["VERSION"]
plugin_fb.close()

# Sorting of plugins in alphabetical order
sorted_base_plugins = {k: plugins[k] for k in sorted(plugins, key=lambda x: (x.startswith('__'), x))}
sorted_base_plugins = {
k: plugins[k] for k in sorted(plugins, key=lambda x: (x.startswith("__"), x))
}
sorted_pluginsscheduled = {k: pluginsscheduled[k] for k in sorted(pluginsscheduled)}

return [sorted_base_plugins, sorted_pluginsscheduled, base_plugin_path]
Expand Down