pre-commit/pre_commit/commands/hook_impl.py at v5 · pre-commit/pre-commit · GitHub

Repository navigation

History

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

from __future__ import annotations

import argparse

import os.path

import subprocess

import sys

from collections.abc import Sequence

from pre_commit.commands.run import run

from pre_commit.envcontext import envcontext

from pre_commit.parse_shebang import normalize_cmd

from pre_commit.store import Store

Z40 = '0' * 40

def _run_legacy(

hook_type: str,

hook_dir: str,

args: Sequence[str],

) -> tuple[int, bytes]:

if os.environ.get('PRE_COMMIT_RUNNING_LEGACY'):

raise SystemExit(

f"bug: pre-commit's script is installed in migration mode\n"

f'run `pre-commit install -f --hook-type {hook_type}` to fix '

f'this\n\n'

f'Please report this bug at '

f'https://github.com/pre-commit/pre-commit/issues',

)

if hook_type == 'pre-push':

stdin = sys.stdin.buffer.read()

else:

stdin = b''

# not running in legacy mode

legacy_hook = os.path.join(hook_dir, f'{hook_type}.legacy')

if not os.access(legacy_hook, os.X_OK):

return 0, stdin

with envcontext((('PRE_COMMIT_RUNNING_LEGACY', '1'),)):

cmd = normalize_cmd((legacy_hook, *args))

return subprocess.run(cmd, input=stdin).returncode, stdin

def _validate_config(

retv: int,

config: str,

skip_on_missing_config: bool,

) -> None:

if not os.path.isfile(config):

if skip_on_missing_config or os.getenv('PRE_COMMIT_ALLOW_NO_CONFIG'):

print(f'`{config}` config file not found. Skipping `pre-commit`.')

raise SystemExit(retv)

else:

print(

f'No {config} file was found\n'

f'- To temporarily silence this, run '

f'`PRE_COMMIT_ALLOW_NO_CONFIG=1 git ...`\n'

f'- To permanently silence this, install pre-commit with the '

f'--allow-missing-config option\n'

f'- To uninstall pre-commit run `pre-commit uninstall`',

)

raise SystemExit(1)

def _ns(

hook_type: str,

color: bool,

all_files: bool = False,

remote_branch: str | None = None,

local_branch: str | None = None,

from_ref: str | None = None,

to_ref: str | None = None,

pre_rebase_upstream: str | None = None,

pre_rebase_branch: str | None = None,

remote_name: str | None = None,

remote_url: str | None = None,

commit_msg_filename: str | None = None,

prepare_commit_message_source: str | None = None,

commit_object_name: str | None = None,

checkout_type: str | None = None,

is_squash_merge: str | None = None,

rewrite_command: str | None = None,

) -> argparse.Namespace:

return argparse.Namespace(

color=color,

hook_stage=hook_type,

remote_branch=remote_branch,

local_branch=local_branch,

from_ref=from_ref,

to_ref=to_ref,

pre_rebase_upstream=pre_rebase_upstream,

pre_rebase_branch=pre_rebase_branch,

remote_name=remote_name,

remote_url=remote_url,

commit_msg_filename=commit_msg_filename,

prepare_commit_message_source=prepare_commit_message_source,

commit_object_name=commit_object_name,

all_files=all_files,

checkout_type=checkout_type,

is_squash_merge=is_squash_merge,

rewrite_command=rewrite_command,

files=(),

hook=None,

verbose=False,

show_diff_on_failure=False,

fail_fast=False,

)

def _rev_exists(rev: str) -> bool:

return not subprocess.call(('git', 'rev-list', '--quiet', rev))

def _pre_push_ns(

color: bool,

args: Sequence[str],

stdin: bytes,

) -> argparse.Namespace | None:

remote_name = args[0]

remote_url = args[1]

for line in stdin.decode().splitlines():

parts = line.rsplit(maxsplit=3)

local_branch, local_sha, remote_branch, remote_sha = parts

if local_sha == Z40:

continue

elif remote_sha != Z40 and _rev_exists(remote_sha):

return _ns(

'pre-push', color,

from_ref=remote_sha, to_ref=local_sha,

remote_branch=remote_branch,

local_branch=local_branch,

remote_name=remote_name, remote_url=remote_url,

)

else:

# ancestors not found in remote

ancestors = subprocess.check_output((

'git', 'rev-list', local_sha, '--topo-order', '--reverse',

'--not', f'--remotes={remote_name}',

)).decode().strip()

if not ancestors:

continue

else:

first_ancestor = ancestors.splitlines()[0]

cmd = ('git', 'rev-list', '--max-parents=0', local_sha)

roots = set(subprocess.check_output(cmd).decode().splitlines())

if first_ancestor in roots:

# pushing the whole tree including root commit

return _ns(

'pre-push', color,

all_files=True,

remote_name=remote_name, remote_url=remote_url,

remote_branch=remote_branch,

local_branch=local_branch,

)

else:

rev_cmd = ('git', 'rev-parse', f'{first_ancestor}^')

source = subprocess.check_output(rev_cmd).decode().strip()

return _ns(

'pre-push', color,

from_ref=source, to_ref=local_sha,

remote_name=remote_name, remote_url=remote_url,

remote_branch=remote_branch,

local_branch=local_branch,

)

# nothing to push

return None

_EXPECTED_ARG_LENGTH_BY_HOOK = {

'commit-msg': 1,

'post-checkout': 3,

'post-commit': 0,

'pre-commit': 0,

'pre-merge-commit': 0,

'post-merge': 1,

'post-rewrite': 1,

'pre-push': 2,

}

def _check_args_length(hook_type: str, args: Sequence[str]) -> None:

if hook_type == 'prepare-commit-msg':

if len(args) < 1 or len(args) > 3:

raise SystemExit(

f'hook-impl for {hook_type} expected 1, 2, or 3 arguments '

f'but got {len(args)}: {args}',

)

elif hook_type == 'pre-rebase':

if len(args) < 1 or len(args) > 2:

raise SystemExit(

f'hook-impl for {hook_type} expected 1 or 2 arguments '

f'but got {len(args)}: {args}',

)

elif hook_type in _EXPECTED_ARG_LENGTH_BY_HOOK:

expected = _EXPECTED_ARG_LENGTH_BY_HOOK[hook_type]

if len(args) != expected:

arguments_s = 'argument' if expected == 1 else 'arguments'

raise SystemExit(

f'hook-impl for {hook_type} expected {expected} {arguments_s} '

f'but got {len(args)}: {args}',

)

else:

raise AssertionError(f'unexpected hook type: {hook_type}')

def _run_ns(

hook_type: str,

color: bool,

args: Sequence[str],

stdin: bytes,

) -> argparse.Namespace | None:

_check_args_length(hook_type, args)

if hook_type == 'pre-push':

return _pre_push_ns(color, args, stdin)

elif hook_type in 'commit-msg':

return _ns(hook_type, color, commit_msg_filename=args[0])

elif hook_type == 'prepare-commit-msg' and len(args) == 1:

return _ns(hook_type, color, commit_msg_filename=args[0])

elif hook_type == 'prepare-commit-msg' and len(args) == 2:

return _ns(

hook_type, color, commit_msg_filename=args[0],

prepare_commit_message_source=args[1],

)

elif hook_type == 'prepare-commit-msg' and len(args) == 3:

return _ns(

hook_type, color, commit_msg_filename=args[0],

prepare_commit_message_source=args[1], commit_object_name=args[2],

)

elif hook_type in {'post-commit', 'pre-merge-commit', 'pre-commit'}:

return _ns(hook_type, color)

elif hook_type == 'post-checkout':

return _ns(

hook_type, color,

from_ref=args[0], to_ref=args[1], checkout_type=args[2],

)

elif hook_type == 'post-merge':

return _ns(hook_type, color, is_squash_merge=args[0])

elif hook_type == 'post-rewrite':

return _ns(hook_type, color, rewrite_command=args[0])

elif hook_type == 'pre-rebase' and len(args) == 1:

return _ns(hook_type, color, pre_rebase_upstream=args[0])

elif hook_type == 'pre-rebase' and len(args) == 2:

return _ns(

hook_type, color, pre_rebase_upstream=args[0],

pre_rebase_branch=args[1],

)

else:

raise AssertionError(f'unexpected hook type: {hook_type}')

def hook_impl(

store: Store,

config: str,

color: bool,

hook_type: str,

hook_dir: str,

skip_on_missing_config: bool,

args: Sequence[str],

) -> int:

retv, stdin = _run_legacy(hook_type, hook_dir, args)

_validate_config(retv, config, skip_on_missing_config)

ns = _run_ns(hook_type, color, args, stdin)

if ns is None:

return retv

else:

return retv | run(config, store, ns)

pre-commit/pre_commit/commands/hook_impl.py at v5 · pre-commit/pre-commit · GitHub

Repository navigation

Expand file tree

Breadcrumbs

hook_impl.py

Latest commit

History

Breadcrumbs

hook_impl.py

File metadata and controls