commoncode/src/commoncode/command.py at main · aboutcode-org/commoncode · GitHub

History

326 lines (258 loc) · 9.36 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

# SPDX-License-Identifier: Apache-2.0

# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.

# See https://github.com/aboutcode-org/commoncode for support or download.

# See https://aboutcode.org for more information about nexB OSS projects.

import contextlib

import ctypes

import io

import logging

import os

import signal

import subprocess

from os import path

from commoncode import text

from commoncode.fileutils import get_temp_dir

from commoncode.system import on_posix

from commoncode.system import on_windows

"""

Wrapper for executing external commands in sub-processes which works

consistently the same way on POSIX and Windows OSes and can cope with the

capture of large command standard outputs without exhausting memory.

"""

logger = logging.getLogger(__name__)

TRACE = False

if TRACE:

import sys

logging.basicConfig(stream=sys.stdout)

logger.setLevel(logging.DEBUG)

# current directory is the root dir of this library

curr_dir = path.dirname(path.dirname(path.abspath(__file__)))

PATH_ENV_VAR = "PATH"

LD_LIBRARY_PATH = "LD_LIBRARY_PATH"

DYLD_LIBRARY_PATH = "DYLD_LIBRARY_PATH"

def execute(cmd_loc, args, cwd=None, env=None, to_files=False, log=TRACE):

"""

Run a `cmd_loc` command with the `args` arguments list and return the return

code, the stdout and stderr.

To avoid RAM exhaustion, always write stdout and stderr streams to files.

If `to_files` is False, return the content of stderr and stdout as ASCII

strings. Otherwise, return the locations to the stderr and stdout temporary

files.

Run the command using the `cwd` current working directory with an `env` dict

of environment variables.

"""

assert cmd_loc

full_cmd = [cmd_loc] + (args or [])

# any shared object should be either in the PATH, the rpath or

# side-by-side with the exceutable

cmd_dir = os.path.dirname(cmd_loc)

env = get_env(env, lib_dir=cmd_dir) or None

cwd = cwd or curr_dir

# temp files for stderr and stdout

tmp_dir = get_temp_dir(prefix="cmd-")

sop = path.join(tmp_dir, "stdout")

sep = path.join(tmp_dir, "stderr")

# shell==True is DANGEROUS but we are not running arbitrary commands

# though we can execute commands that just happen to be in the path

# See why we need it on Windows https://bugs.python.org/issue8557

shell = True if on_windows else False

if log:

printer = logger.debug if TRACE else lambda x: print(x)

printer(

"Executing command %(cmd_loc)r as:\n%(full_cmd)r\nwith: env=%(env)r\n"

"shell=%(shell)r\ncwd=%(cwd)r\nstdout=%(sop)r\nstderr=%(sep)r" % locals()

)

proc = None

rc = 100

try:

with io.open(sop, "wb") as stdout, io.open(sep, "wb") as stderr, pushd(cmd_dir):

proc = subprocess.Popen(

full_cmd,

cwd=cwd,

env=env,

stdout=stdout,

stderr=stderr,

shell=shell,

# -1 defaults bufsize to system bufsize

bufsize=-1,

universal_newlines=True,

)

stdout, stderr = proc.communicate()

rc = proc.returncode if proc else 0

finally:

close(proc)

if not to_files:

# return output as ASCII string loaded from the output files

with open(sop, "rb") as so:

sor = so.read()

sop = text.toascii(sor).strip()

with open(sep, "rb") as se:

ser = se.read()

sep = text.toascii(ser).strip()

return rc, sop, sep

def execute2(

cmd_loc,

args,

lib_dir=None,

cwd=None,

env=None,

to_files=False,

log=TRACE,

"""

Run a `cmd_loc` command with the `args` arguments list and return the return

code, the stdout and stderr.

DEPRECATED: DO NOT USE. Use execute() instead

To avoid RAM exhaustion, always write stdout and stderr streams to files.

If `to_files` is False, return the content of stderr and stdout as ASCII

strings. Otherwise, return the locations to the stderr and stdout temporary

files.

Run the command using the `cwd` current working directory with an `env` dict

of environment variables.

"""

import warnings

warnings.warn(

"commoncode.command.execute2 is deprecated. Use execute() instead.",

DeprecationWarning,

stacklevel=2,

)

return execute(cmd_loc, args, cwd, env, to_files, log)

def get_env(base_vars=None, lib_dir=None):

"""

Return a dictionary of environment variables for command execution with

appropriate DY/LD_LIBRARY_PATH path variables. Use the optional `base_vars`

environment variables dictionary as a base if provided. Note: if `base_vars`

contains DY/LD_LIBRARY_PATH variables these will be overwritten. On POSIX,

add `lib_dir` as DY/LD_LIBRARY_PATH-like path if provided.

"""

env_vars = {}

if base_vars:

env_vars.update(base_vars)

# Create and add LD environment variables

if lib_dir and on_posix:

new_path = f"{lib_dir}"

# on Linux/posix

ld_lib_path = os.environ.get(LD_LIBRARY_PATH)

env_vars.update({LD_LIBRARY_PATH: update_path_var(ld_lib_path, new_path)})

# on Mac, though LD_LIBRARY_PATH should work too

dyld_lib_path = os.environ.get(DYLD_LIBRARY_PATH)

env_vars.update({DYLD_LIBRARY_PATH: update_path_var(dyld_lib_path, new_path)})

env_vars = {text.as_unicode(k): text.as_unicode(v) for k, v in env_vars.items()}

return env_vars

def close(proc):

"""

Close a `proc` process opened pipes and kill the process.

"""

if not proc:

return

def close_pipe(p):

if not p:

return

try:

p.close()

except IOError:

pass

close_pipe(getattr(proc, "stdin", None))

close_pipe(getattr(proc, "stdout", None))

close_pipe(getattr(proc, "stderr", None))

try:

# Ensure process death otherwise proc.wait may hang in some cases

# NB: this will run only on POSIX OSes supporting signals

os.kill(proc.pid, signal.SIGKILL) # NOQA

except Exception:

pass

# This may slow things down a tad on non-POSIX Oses but is safe:

# this calls os.waitpid() to make sure the process is dead

proc.wait()

def load_shared_library(dll_loc, *args):

"""

Return the loaded shared library object from the ``dll_loc`` location.

"""

if not dll_loc or not path.exists(dll_loc):

raise ImportError(f"Shared library does not exists: dll_loc: {dll_loc}")

if not isinstance(dll_loc, str):

dll_loc = os.fsdecode(dll_loc)

lib = None

dll_dir = os.path.dirname(dll_loc)

try:

with pushd(dll_dir):

lib = ctypes.CDLL(dll_loc)

except OSError as e:

import traceback

from pprint import pformat

msgs = tuple(

[

f'ctypes.CDLL("{dll_loc}")',

"os.environ:\n{}".format(pformat(dict(os.environ))),

traceback.format_exc(),

]

)

raise Exception(msgs) from e

if lib and lib._name:

return lib

raise Exception(f"Failed to load shared library with ctypes: {dll_loc}")

@contextlib.contextmanager

def pushd(path):

"""

Context manager to change the current working directory to `path`.

"""

original_cwd = os.getcwd()

if not path:

path = original_cwd

try:

os.chdir(path)

yield os.getcwd()

finally:

os.chdir(original_cwd)

def update_path_var(existing_path_var, new_path):

"""

Return an updated value for the `existing_path_var` PATH-like environment

variable value by adding `new_path` to the front of that variable if

`new_path` is not already part of this PATH-like variable.

"""

if not new_path:

return existing_path_var

existing_path_var = existing_path_var or ""

existing_path_var = os.fsdecode(existing_path_var)

new_path = os.fsdecode(new_path)

path_elements = existing_path_var.split(os.pathsep)

if not path_elements:

updated_path_var = new_path

elif new_path not in path_elements:

# add new path to the front of the PATH env var

path_elements.insert(0, new_path)

updated_path_var = os.pathsep.join(path_elements)

else:

# new path is already in PATH, change nothing

updated_path_var = existing_path_var

if not isinstance(updated_path_var, str):

updated_path_var = os.fsdecode(updated_path_var)

return updated_path_var

PATH_VARS = (

DYLD_LIBRARY_PATH,

LD_LIBRARY_PATH,

"PATH",

)

def searchable_paths(env_vars=PATH_VARS):

"""

Return a list of directories where to search "in the PATH" in the provided

``env_vars`` list of PATH-like environment variables.

"""

dirs = []

for env_var in env_vars:

value = os.environ.get(env_var, "") or ""

dirs.extend(value.split(os.pathsep))

dirs = [os.path.realpath(d.strip()) for d in dirs if d.strip()]

return tuple(d for d in dirs if os.path.isdir(d))

def find_in_path(filename, searchable_paths=searchable_paths()):

"""

Return the location of a ``filename`` found in the ``searchable_paths`` list

of directory or None.

"""

for path in searchable_paths:

location = os.path.join(path, filename)

if os.path.exists(location):

return location

commoncode/src/commoncode/command.py at main · aboutcode-org/commoncode · GitHub

Uh oh!

Uh oh!

commoncode/src/commoncode/command.py at main · aboutcode-org/commoncode · GitHub

Uh oh!

Uh oh!

Expand file tree

Breadcrumbs

command.py

Latest commit

History

Breadcrumbs

command.py

File metadata and controls