forked from colbymchenry/codegraph
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcelery-dispatch-synthesizer.test.ts
More file actions
129 lines (112 loc) · 4.71 KB
/
Copy pathcelery-dispatch-synthesizer.test.ts
File metadata and controls
129 lines (112 loc) · 4.71 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
/**
* Celery task-dispatch bridge (Python).
*
* Celery decouples a task's call site from its body: a `@shared_task` / `@app.task`
* decorated `def` is invoked through `task.delay(...)` / `task.apply_async(...)`, a
* dynamic hop with no static edge. This bridges each `.delay`/`.apply_async` site to
* the task function, gated on the DECORATOR (read from the source above the `def`) so a
* `.delay()` on a non-task object resolves to nothing. Covers both decorator dialects
* (`@shared_task`, `@app.task(...)`), the module-qualified `mod.task.apply_async()` form,
* and proves the precision gates: a plain function called with `.delay()` and a canvas
* `group(...).delay()` (no single identifier before `.delay`) both contribute no edge.
*/
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import * as fs from 'node:fs';
import * as path from 'node:path';
import * as os from 'node:os';
import { CodeGraph } from '../src';
describe('celery-dispatch synthesizer', () => {
let dir: string;
beforeEach(() => { dir = fs.mkdtempSync(path.join(os.tmpdir(), 'celery-dispatch-')); });
afterEach(() => { fs.rmSync(dir, { recursive: true, force: true }); });
it('bridges .delay()/.apply_async() to decorated tasks, ignoring non-task and canvas dispatch', async () => {
// Two decorator dialects: bare @shared_task and arg'd @app.task(...).
fs.writeFileSync(
path.join(dir, 'tasks.py'),
`from celery import shared_task
from myapp.celery import app
@shared_task
def send_email(to):
return to
@app.task(bind=True, max_retries=3)
def crunch(self, n):
return n * 2
`
);
fs.mkdirSync(path.join(dir, 'services'), { recursive: true });
fs.writeFileSync(
path.join(dir, 'services', 'tickets.py'),
`from celery import shared_task
@shared_task
def invalidate_cache():
return None
`
);
// A plain function — NOT a celery task — that nonetheless has .delay() called on it.
fs.writeFileSync(
path.join(dir, 'utils.py'),
`def process_data(x):
return x
`
);
// Dispatch sites, all inside one enclosing function.
fs.writeFileSync(
path.join(dir, 'views.py'),
`from tasks import send_email, crunch
from services import tickets
from utils import process_data
from celery import group
def handle_request(req):
send_email.delay(req.addr) # → send_email task (cross-file)
crunch.apply_async(args=[5]) # → crunch task (@app.task dialect)
tickets.invalidate_cache.apply_async() # module-qualified → invalidate_cache
process_data.delay(req.x) # NOT a task → no edge
group([send_email.s(a) for a in req.addrs]).delay() # canvas → no edge
`
);
const cg = await CodeGraph.init(dir, { silent: true });
await cg.indexAll();
const db = (cg as any).db.db;
const edges = db
.prepare(
`SELECT s.name source, t.name target, t.file_path tf, json_extract(e.metadata,'$.via') via
FROM edges e JOIN nodes s ON s.id = e.source JOIN nodes t ON t.id = e.target
WHERE json_extract(e.metadata,'$.synthesizedBy') = 'celery-dispatch'`
)
.all();
const targets = (src: string) => edges.filter((r: any) => r.source === src).map((r: any) => r.target).sort();
// handle_request dispatches exactly the three real tasks (both dialects + module-qualified).
expect(targets('handle_request')).toEqual(['crunch', 'invalidate_cache', 'send_email']);
// The @app.task target resolved to the task def, not anything else.
const crunchEdge = edges.find((r: any) => r.target === 'crunch');
expect(crunchEdge.tf).toMatch(/tasks\.py$/);
// Module-qualified `tickets.invalidate_cache.apply_async()` resolved by the last identifier.
const cacheEdge = edges.find((r: any) => r.target === 'invalidate_cache');
expect(cacheEdge.tf).toMatch(/services[\\/]tickets\.py$/);
expect(cacheEdge.via).toBe('invalidate_cache');
// PRECISION: a plain function called with .delay() is never targeted (no decorator).
expect(edges.some((r: any) => r.target === 'process_data')).toBe(false);
cg.close?.();
});
it('produces no edges in a Celery-free project (clean control)', async () => {
fs.writeFileSync(
path.join(dir, 'app.py'),
`def schedule(job):
job.delay() # a .delay() that has nothing to do with Celery
return job
def run():
schedule(make_job())
`
);
const cg = await CodeGraph.init(dir, { silent: true });
await cg.indexAll();
const db = (cg as any).db.db;
const count = db
.prepare(
`SELECT count(*) c FROM edges WHERE json_extract(metadata,'$.synthesizedBy') = 'celery-dispatch'`
)
.get();
expect(count.c).toBe(0);
cg.close?.();
});
});